events project, spotify mapping, lastfm mapping, CLI reacting to db changes

Selector.CLI/DbWatcherService.cs

@@ -1,20 +1,17 @@
 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 Selector.Events;
-using StackExchange.Redis;
+using System.Collections.Concurrent;
 
 namespace Selector.CLI
 {
@@ -24,6 +21,7 @@ namespace Selector.CLI
 
         private readonly ILogger<DbWatcherService> Logger;
         private readonly IServiceProvider ServiceProvider;
+        private readonly UserEventBus UserEventBus;
 
         private readonly IWatcherFactory WatcherFactory;
         private readonly IWatcherCollectionFactory WatcherCollectionFactory;
@@ -34,7 +32,7 @@ namespace Selector.CLI
 
         private readonly IPublisherFactory PublisherFactory;
         private readonly ICacheWriterFactory CacheWriterFactory;
-        private Dictionary<string, IWatcherCollection> Watchers { get; set; } = new();
+        private ConcurrentDictionary<string, IWatcherCollection> Watchers { get; set; } = new();
 
         public DbWatcherService(
             IWatcherFactory watcherFactory,
@@ -44,6 +42,7 @@ namespace Selector.CLI
             IAudioFeatureInjectorFactory audioFeatureInjectorFactory,
             IPlayCounterFactory playCounterFactory,
 
+            UserEventBus userEventBus,
 
             ILogger<DbWatcherService> logger,
             IServiceProvider serviceProvider,
@@ -54,6 +53,7 @@ namespace Selector.CLI
         {
             Logger = logger;
             ServiceProvider = serviceProvider;
+            UserEventBus = userEventBus;
 
             WatcherFactory = watcherFactory;
             WatcherCollectionFactory = watcherCollectionFactory;
@@ -71,6 +71,7 @@ namespace Selector.CLI
             Logger.LogInformation("Starting database watcher service...");
 
             var watcherIndices = await InitInstances();
+            AttachEventBus();
 
             Logger.LogInformation("Starting {count} affected watcher collection(s)...", watcherIndices.Count());
             StartWatcherCollections(watcherIndices);
@@ -87,53 +88,60 @@ namespace Selector.CLI
                                         .Include(w => w.User)
                                         .Where(w => !string.IsNullOrWhiteSpace(w.User.SpotifyRefreshToken)))
             {
-                Logger.LogInformation("Creating new [{type}] watcher", dbWatcher.Type);
-
                 var watcherCollectionIdx = dbWatcher.UserId;
                 indices.Add(watcherCollectionIdx);
 
-                if (!Watchers.ContainsKey(watcherCollectionIdx))
+                await InitInstance(dbWatcher);
-                    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);
-
-                        consumers.Add(await AudioFeatureInjectorFactory.Get(spotifyFactory));
-                        if (CacheWriterFactory is not null) consumers.Add(await CacheWriterFactory.Get());
-                        if (PublisherFactory is not null) consumers.Add(await PublisherFactory.Get());
-
-                        if (!string.IsNullOrWhiteSpace(dbWatcher.User.LastFmUsername))
-                        {
-                            consumers.Add(await PlayCounterFactory.Get(creds: new() { Username = dbWatcher.User.LastFmUsername }));
-                        }
-                        else
-                        {
-                            Logger.LogDebug("[{username}] No Last.fm username, skipping play counter", dbWatcher.User.UserName);
-                        }
-
-                        break;
-
-                    case WatcherType.Playlist:
-                        throw new NotImplementedException("Playlist watchers not implemented");
-                        // break;
-                }
-
-                watcherCollection.Add(watcher, consumers);
             }
 
             return indices;
         }
 
+        private async Task<IWatcherContext> InitInstance(Watcher dbWatcher)
+        {
+            Logger.LogInformation("Creating new [{type}] watcher", dbWatcher.Type);
+
+            var watcherCollectionIdx = dbWatcher.UserId;
+
+            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);
+
+                    consumers.Add(await AudioFeatureInjectorFactory.Get(spotifyFactory));
+                    if (CacheWriterFactory is not null) consumers.Add(await CacheWriterFactory.Get());
+                    if (PublisherFactory is not null) consumers.Add(await PublisherFactory.Get());
+
+                    if (!string.IsNullOrWhiteSpace(dbWatcher.User.LastFmUsername))
+                    {
+                        consumers.Add(await PlayCounterFactory.Get(creds: new() { Username = dbWatcher.User.LastFmUsername }));
+                    }
+                    else
+                    {
+                        Logger.LogDebug("[{username}] No Last.fm username, skipping play counter", dbWatcher.User.UserName);
+                    }
+
+                    break;
+
+                case WatcherType.Playlist:
+                    throw new NotImplementedException("Playlist watchers not implemented");
+                    // break;
+            }
+
+            return watcherCollection.Add(watcher, consumers);
+        }
+
         private void StartWatcherCollections(IEnumerable<string> indices)
         {
             foreach (var index in indices)
@@ -160,7 +168,81 @@ namespace Selector.CLI
                 watcher.Stop();
             }
 
+            DetachEventBus();
+
             return Task.CompletedTask;
         }
+
+        private void AttachEventBus()
+        {
+            UserEventBus.SpotifyLinkChange += SpotifyChangeCallback;
+            UserEventBus.LastfmCredChange += LastfmChangeCallback;
+        }
+
+        private void DetachEventBus()
+        {
+            UserEventBus.SpotifyLinkChange -= SpotifyChangeCallback;
+            UserEventBus.LastfmCredChange -= LastfmChangeCallback;
+        }
+
+        public async void SpotifyChangeCallback(object sender, SpotifyLinkChange change)
+        {
+            if(Watchers.ContainsKey(change.UserId))
+            {
+                Logger.LogDebug("Setting new Spotify link state for [{username}], [{}]", change.UserId, change.NewLinkState);
+
+                var watcherCollection = Watchers[change.UserId];
+
+                if(change.NewLinkState)
+                {
+                    watcherCollection.Start();
+                }
+                else
+                {
+                    watcherCollection.Stop();
+                }
+            }
+            else
+            {
+                using var scope = ServiceProvider.CreateScope();
+                var db = scope.ServiceProvider.GetService<ApplicationDbContext>();
+
+                var watcherEnum = db.Watcher
+                                    .Include(w => w.User)
+                                    .Where(w => w.UserId == change.UserId);
+
+                foreach (var dbWatcher in watcherEnum)
+                {
+                    var context = await InitInstance(dbWatcher);
+                }
+
+                Watchers[change.UserId].Start();
+
+                Logger.LogDebug("Started {} watchers for [{username}]", watcherEnum.Count(), change.UserId);
+            }
+        }
+
+        public void LastfmChangeCallback(object sender, LastfmChange change)
+        {
+            if (Watchers.ContainsKey(change.UserId))
+            {
+                Logger.LogDebug("Setting new username for [{}], [{}]", change.UserId, change.NewUsername);
+
+                var watcherCollection = Watchers[change.UserId];
+
+                foreach(var watcher in watcherCollection.Consumers)
+                {
+                    if(watcher is PlayCounter counter)
+                    {
+                        counter.Credentials.Username = change.NewUsername;
+                    }
+                }
+            }
+            else
+            {
+
+                Logger.LogDebug("No watchers running for [{username}], skipping Spotify event", change.UserId);
+            }
+        }
     }
 }

Selector.CLI/Program.cs

@@ -11,9 +11,7 @@ using Selector.Extensions;
 using Selector.Model;
 using Selector.Cache;
 using Selector.Cache.Extensions;
-
+using Selector.Events;
-using IF.Lastfm.Core.Api;
-using Selector.Model.Extensions;
 
 namespace Selector.CLI
 {
@@ -114,19 +112,25 @@ namespace Selector.CLI
             }
             services.AddWatcher();
 
-            services.AddSpotify();
+            services.AddEvents();
-            if (config.RedisOptions.Enabled) {
-                Console.WriteLine("> Adding caching Spotify consumers...");
-                services.AddCachingSpotify(); 
-            }
 
+            services.AddSpotify();
             ConfigureLastFm(config, services);
             ConfigureDb(config, services);
+            ConfigureEqual(config, services);
 
-            if (config.RedisOptions.Enabled) 
+            if (config.RedisOptions.Enabled)
+            {
+                Console.WriteLine("> Adding Redis...");
                 services.AddRedisServices(config.RedisOptions.ConnectionString);
 
-            ConfigureEqual(config, services);
+                Console.WriteLine("> Adding cache event maps...");
+                services.AddTransient<IEventMapping, SpotifyLinkFromCacheMapping>();
+                services.AddTransient<IEventMapping, LastfmFromCacheMapping>();
+
+                Console.WriteLine("> Adding caching Spotify consumers...");
+                services.AddCachingSpotify();
+            }
 
             // HOSTED SERVICES
             if (config.WatcherOptions.Enabled)

Selector.CLI/Selector.CLI.csproj

@@ -25,6 +25,7 @@
     <ProjectReference Include="..\Selector\Selector.csproj" />
     <ProjectReference Include="..\Selector.Model\Selector.Model.csproj" />
     <ProjectReference Include="..\Selector.Cache\Selector.Cache.csproj" />
+    <ProjectReference Include="..\Selector.Event\Selector.Event.csproj" />
   </ItemGroup>
 
   <ItemGroup>

Selector.Event/CacheMappings/IEventMapping.cs

@@ -4,9 +4,9 @@ using System.Threading.Tasks;
 using Microsoft.AspNetCore.SignalR;
 using StackExchange.Redis;
 
-namespace Selector.Web.Service
+namespace Selector.Events
 {
-    public interface ICacheEventMapping
+    public interface IEventMapping
     {
         public Task ConstructMapping();
     }

Selector.Event/CacheMappings/LastfmMapping.cs

@@ -0,0 +1,87 @@
+using System.Text.Json;
+using Microsoft.Extensions.Logging;
+
+using StackExchange.Redis;
+
+using Selector.Cache;
+
+namespace Selector.Events
+{
+    public class LastfmChange
+    {
+        public string UserId { get; set; }
+        public string PreviousUsername { get; set; }
+        public string NewUsername { get; set; }
+    }
+
+    public class LastfmFromCacheMapping : IEventMapping
+    {
+        private readonly ILogger<LastfmFromCacheMapping> Logger;
+        private readonly ISubscriber Subscriber;
+        private readonly UserEventBus UserEvent;
+
+        public LastfmFromCacheMapping(ILogger<LastfmFromCacheMapping> logger, 
+            ISubscriber subscriber,
+            UserEventBus userEvent)
+        {
+            Logger = logger;
+            Subscriber = subscriber;
+            UserEvent = userEvent;
+        }
+
+        public async Task ConstructMapping()
+        {
+            Logger.LogDebug("Forming Last.fm username event mapping FROM cache TO event bus");
+
+            (await Subscriber.SubscribeAsync(Key.AllUserLastfm)).OnMessage(message => {
+                
+                try{
+                    var userId = Key.Param(message.Channel);
+
+                    var deserialised = JsonSerializer.Deserialize<LastfmChange>(message.Message);
+                    Logger.LogDebug("Received new Last.fm username event for [{userId}]", deserialised.UserId);
+
+                    if (!userId.Equals(deserialised.UserId))
+                    {
+                        Logger.LogWarning("Serialised user ID [{}] does not match cache channel [{}]", userId, deserialised.UserId);
+                    }
+
+                    UserEvent.OnLastfmCredChange(this, deserialised);
+                }
+                catch(Exception e)
+                {
+                    Logger.LogError(e, "Error parsing Last.fm username event");
+                }
+            });
+        }
+    }
+
+    public class LastfmToCacheMapping : IEventMapping
+    {
+        private readonly ILogger<LastfmToCacheMapping> Logger;
+        private readonly ISubscriber Subscriber;
+        private readonly UserEventBus UserEvent;
+
+        public LastfmToCacheMapping(ILogger<LastfmToCacheMapping> logger,
+            ISubscriber subscriber,
+            UserEventBus userEvent)
+        {
+            Logger = logger;
+            Subscriber = subscriber;
+            UserEvent = userEvent;
+        }
+
+        public Task ConstructMapping()
+        {
+            Logger.LogDebug("Forming Last.fm username event mapping TO cache FROM event bus");
+
+            UserEvent.LastfmCredChange += async (o, e) =>
+            {
+                var payload = JsonSerializer.Serialize(e);
+                await Subscriber.PublishAsync(Key.UserLastfm(e.UserId), payload);
+            };
+
+            return Task.CompletedTask;
+        }
+    }
+}

Selector.Event/CacheMappings/NowPlayingMapping.cs

@@ -1,25 +1,19 @@
-using System;
 using System.Text.Json;
-using System.Linq;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Logging;
 
 using StackExchange.Redis;
 
-using Selector.Web.Hubs;
 using Selector.Cache;
-using Selector.Model.Events;
 
-namespace Selector.Web.Service
+namespace Selector.Events
 {
-    public class NowPlayingCacheMapping : ICacheEventMapping
+    public class NowPlayingFromCacheMapping : IEventMapping
     {
-        private readonly ILogger<NowPlayingCacheMapping> Logger;
+        private readonly ILogger<NowPlayingFromCacheMapping> Logger;
         private readonly ISubscriber Subscriber;
         private readonly UserEventBus UserEvent;
 
-        public NowPlayingCacheMapping(ILogger<NowPlayingCacheMapping> logger, 
+        public NowPlayingFromCacheMapping(ILogger<NowPlayingFromCacheMapping> logger, 
             ISubscriber subscriber,
             UserEventBus userEvent)
         {

Selector.Event/CacheMappings/SpotifyMapping.cs

@@ -0,0 +1,91 @@
+using System.Text.Json;
+using Microsoft.Extensions.Logging;
+
+using StackExchange.Redis;
+
+using Selector.Cache;
+
+namespace Selector.Events
+{
+    public class SpotifyLinkChange
+    {
+        public string UserId { get; set; }
+        public bool PreviousLinkState { get; set; }
+        public bool NewLinkState { get; set; }
+    }
+
+    public class SpotifyLinkFromCacheMapping : IEventMapping
+    {
+        private readonly ILogger<SpotifyLinkFromCacheMapping> Logger;
+        private readonly ISubscriber Subscriber;
+        private readonly UserEventBus UserEvent;
+
+        public SpotifyLinkFromCacheMapping(ILogger<SpotifyLinkFromCacheMapping> logger, 
+            ISubscriber subscriber,
+            UserEventBus userEvent)
+        {
+            Logger = logger;
+            Subscriber = subscriber;
+            UserEvent = userEvent;
+        }
+
+        public async Task ConstructMapping()
+        {
+            Logger.LogDebug("Forming Spotify link event mapping FROM cache TO event bus");
+
+            (await Subscriber.SubscribeAsync(Key.AllUserSpotify)).OnMessage(message => {
+                
+                try{
+                    var userId = Key.Param(message.Channel);
+
+                    var deserialised = JsonSerializer.Deserialize<SpotifyLinkChange>(message.Message);
+                    Logger.LogDebug("Received new Spotify link event for [{userId}]", deserialised.UserId);
+
+                    if (!userId.Equals(deserialised.UserId))
+                    {
+                        Logger.LogWarning("Serialised user ID [{}] does not match cache channel [{}]", userId, deserialised.UserId);
+                    }
+
+                    UserEvent.OnSpotifyLinkChange(this, deserialised);
+                }
+                catch(TaskCanceledException)
+                {
+                    Logger.LogDebug("Task Cancelled");
+                }
+                catch(Exception e)
+                {
+                    Logger.LogError(e, "Error parsing new Spotify link event");
+                }
+            });
+        }
+    }
+
+    public class SpotifyLinkToCacheMapping : IEventMapping
+    {
+        private readonly ILogger<SpotifyLinkToCacheMapping> Logger;
+        private readonly ISubscriber Subscriber;
+        private readonly UserEventBus UserEvent;
+
+        public SpotifyLinkToCacheMapping(ILogger<SpotifyLinkToCacheMapping> logger,
+            ISubscriber subscriber,
+            UserEventBus userEvent)
+        {
+            Logger = logger;
+            Subscriber = subscriber;
+            UserEvent = userEvent;
+        }
+
+        public Task ConstructMapping()
+        {
+            Logger.LogDebug("Forming Spotify link event mapping TO cache FROM event bus");
+
+            UserEvent.SpotifyLinkChange += async (o, e) =>
+            {
+                var payload = JsonSerializer.Serialize(e);
+                await Subscriber.PublishAsync(Key.UserSpotify(e.UserId), payload);
+            };
+
+            return Task.CompletedTask;
+        }
+    }
+}

Selector.Event/EventMappingService.cs

@@ -0,0 +1,44 @@
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+
+namespace Selector.Events
+{
+    public class EventMappingService: IHostedService
+    {
+        private readonly ILogger<EventMappingService> Logger;
+        private readonly IServiceScopeFactory ScopeFactory;
+
+        private readonly IEnumerable<IEventMapping> CacheEvents;
+
+        public EventMappingService(
+            ILogger<EventMappingService> logger,
+            IEnumerable<IEventMapping> mappings,
+            IServiceScopeFactory scopeFactory
+        )
+        {
+            Logger = logger;
+            ScopeFactory = scopeFactory;
+
+            CacheEvents = mappings;
+        }
+
+        public Task StartAsync(CancellationToken cancellationToken)
+        {
+            Logger.LogInformation("Starting event mapping service");
+
+            foreach (var mapping in CacheEvents)
+            {
+                mapping.ConstructMapping();
+            }
+
+            return Task.CompletedTask;
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            return Task.CompletedTask;
+        }
+    }
+}

Selector.Event/Extensions/ServiceExtensions.cs

@@ -0,0 +1,24 @@
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Selector.Events
+{
+    public static class ServiceExtensions
+    {
+        public static void AddEvents(this IServiceCollection services)
+        {
+            services.AddEventBus();
+            services.AddEventMappingAgent();
+        }
+
+        public static void AddEventBus(this IServiceCollection services)
+        {
+            services.AddSingleton<UserEventBus>();
+            services.AddSingleton<IEventBus, UserEventBus>(sp => sp.GetRequiredService<UserEventBus>());
+        }
+
+        public static void AddEventMappingAgent(this IServiceCollection services)
+        {
+            services.AddHostedService<EventMappingService>();
+        }
+    }
+}

Selector.Event/Selector.Event.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net6.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Selector\Selector.csproj" />
+    <ProjectReference Include="..\Selector.Model\Selector.Model.csproj" />
+    <ProjectReference Include="..\Selector.Cache\Selector.Cache.csproj" />
+  </ItemGroup>
+
+</Project>

Selector.Event/UserEventBus.cs

@@ -4,17 +4,18 @@ using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
-using Selector.Events;
 
-namespace Selector.Model.Events
+using Selector.Model;
+
+namespace Selector.Events
 {
     public class UserEventBus: IEventBus
     {
         private readonly ILogger<UserEventBus> Logger;
 
         public event EventHandler<ApplicationUser> UserChange;
-        public event EventHandler<ApplicationUser> SpotifyLinkChange;
+        public event EventHandler<SpotifyLinkChange> SpotifyLinkChange;
-        public event EventHandler<ApplicationUser> LastfmCredChange;
+        public event EventHandler<LastfmChange> LastfmCredChange;
 
         public event EventHandler<(string, CurrentlyPlayingDTO)> CurrentlyPlaying;
 
@@ -29,15 +30,15 @@ namespace Selector.Model.Events
             UserChange?.Invoke(sender, args);
         }
 
-        public void OnSpotifyLinkChange(object sender, ApplicationUser args)
+        public void OnSpotifyLinkChange(object sender, SpotifyLinkChange args)
         {
-            Logger.LogTrace("Firing user Spotify event [{usernamne}]", args?.UserName);
+            Logger.LogTrace("Firing user Spotify event [{usernamne}]", args?.UserId);
             SpotifyLinkChange?.Invoke(sender, args);
         }
 
-        public void OnLastfmCredChange(object sender, ApplicationUser args)
+        public void OnLastfmCredChange(object sender, LastfmChange args)
         {
-            Logger.LogTrace("Firing user Last.fm event [{usernamne}]", args?.UserName);
+            Logger.LogTrace("Firing user Last.fm event [{usernamne}]", args?.UserId);
             LastfmCredChange?.Invoke(sender, args);
         }
 

Selector.Model/Extensions/ServiceExtensions.cs

@@ -1,20 +1,12 @@
 using Microsoft.AspNetCore.Authorization;
 using Microsoft.Extensions.DependencyInjection;
 
-using Selector.Events;
 using Selector.Model.Authorisation;
-using Selector.Model.Events;
 
 namespace Selector.Model.Extensions
 {
     public static class ServiceExtensions
     {
-        public static void AddModelEventBus(this IServiceCollection services)
-        {
-            services.AddSingleton<UserEventBus>();
-            services.AddSingleton<IEventBus, UserEventBus>(sp => sp.GetService<UserEventBus>());
-        }
-
         public static void AddAuthorisationHandlers(this IServiceCollection services)
         {
             services.AddAuthorization(options =>

Selector.Web/Areas/Identity/Pages/Account/Manage/Lastfm.cshtml.cs

@@ -12,17 +12,21 @@ using Microsoft.AspNetCore.Mvc.RazorPages;
 using Microsoft.AspNetCore.WebUtilities;
 
 using Selector.Model;
+using Selector.Events;
 
 namespace Selector.Web.Areas.Identity.Pages.Account.Manage
 {
     public partial class LastFmModel : PageModel
     {
         private readonly UserManager<ApplicationUser> _userManager;
+        private readonly UserEventBus UserEvent;
 
         public LastFmModel(
-            UserManager<ApplicationUser> userManager)
+            UserManager<ApplicationUser> userManager,
+            UserEventBus userEvent)
         {
             _userManager = userManager;
+            UserEvent = userEvent;
         }
 
         [TempData]
@@ -75,8 +79,15 @@ namespace Selector.Web.Areas.Identity.Pages.Account.Manage
 
             if (Input.Username != user.LastFmUsername)
             {
+                var oldUsername = user.LastFmUsername;
                 user.LastFmUsername = Input.Username?.Trim();
+                
                 await _userManager.UpdateAsync(user);
+                UserEvent.OnLastfmCredChange(this, new LastfmChange { 
+                    UserId = user.Id, 
+                    PreviousUsername = oldUsername, 
+                    NewUsername = user.LastFmUsername
+                });
 
                 StatusMessage = "Username changed";
                 return RedirectToPage();

Selector.Web/Areas/Identity/Pages/Account/Manage/Spotify.cshtml.cs

@@ -1,21 +1,16 @@
 using System;
-using System.Collections.Generic;
-using System.ComponentModel.DataAnnotations;
-using System.Text;
-using System.Text.Encodings.Web;
-using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Identity;
-using Microsoft.AspNetCore.Identity.UI.Services;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.AspNetCore.Mvc.RazorPages;
-using Microsoft.AspNetCore.WebUtilities;
-
-using Selector.Model;
-using SpotifyAPI.Web;
 using Microsoft.Extensions.Options;
 using Microsoft.Extensions.Logging;
 
+using SpotifyAPI.Web;
+
+using Selector.Model;
+using Selector.Events;
+
 namespace Selector.Web.Areas.Identity.Pages.Account.Manage
 {
     public partial class SpotifyModel : PageModel
@@ -23,16 +18,20 @@ namespace Selector.Web.Areas.Identity.Pages.Account.Manage
         private readonly UserManager<ApplicationUser> _userManager;
         private readonly ILogger<SpotifyModel> Logger;
         private readonly RootOptions Config;
+        private readonly UserEventBus UserEvent;
 
         public SpotifyModel(
             UserManager<ApplicationUser> userManager,
             ILogger<SpotifyModel> logger,
-            IOptions<RootOptions> config
+            IOptions<RootOptions> config,
+            UserEventBus userEvent
             )
         {
             _userManager = userManager;
             Logger = logger;
             Config = config.Value;
+
+            UserEvent = userEvent;
         }
 
         [BindProperty]
@@ -97,8 +96,6 @@ namespace Selector.Web.Areas.Identity.Pages.Account.Manage
                 return NotFound($"Unable to load user with ID '{_userManager.GetUserId(User)}'.");
             }
 
-            // TODO: stop users Spotify-linked resources (watchers)
-
             user.SpotifyIsLinked = false;
 
             user.SpotifyAccessToken = null;
@@ -107,6 +104,7 @@ namespace Selector.Web.Areas.Identity.Pages.Account.Manage
             user.SpotifyLastRefresh = default;
 
             await _userManager.UpdateAsync(user);
+            UserEvent.OnSpotifyLinkChange(this, new SpotifyLinkChange { UserId = user.Id, PreviousLinkState = true, NewLinkState = false });
 
             StatusMessage = "Spotify Unlinked";
             return RedirectToPage();

Selector.Web/Controller/SpotifyController.cs

@@ -8,10 +8,11 @@ using Microsoft.AspNetCore.Authorization;
 using Microsoft.Extensions.Options;
 using Microsoft.Extensions.Logging;
 
-using Selector.Model;
-
 using SpotifyAPI.Web;
 
+using Selector.Events;
+using Selector.Model;
+
 namespace Selector.Web.Controller
 {
 
@@ -20,6 +21,7 @@ namespace Selector.Web.Controller
     public class SpotifyController : BaseAuthController
     {
         private readonly RootOptions Config;
+        private readonly UserEventBus UserEvent;
         private const string ManageSpotifyPath = "/Identity/Account/Manage/Spotify";
 
         public SpotifyController(
@@ -27,10 +29,12 @@ namespace Selector.Web.Controller
             IAuthorizationService auth,
             UserManager<ApplicationUser> userManager,
             ILogger<UsersController> logger,
-            IOptions<RootOptions> config
+            IOptions<RootOptions> config,
+            UserEventBus userEvent
         ) : base(context, auth, userManager, logger) 
         {
             Config = config.Value;
+            UserEvent = userEvent;
         }
 
         [HttpGet]
@@ -76,6 +80,8 @@ namespace Selector.Web.Controller
 
             await UserManager.UpdateAsync(user);
 
+            UserEvent.OnSpotifyLinkChange(this, new SpotifyLinkChange { UserId = user.Id, PreviousLinkState = false, NewLinkState = true });
+
             TempData["StatusMessage"] = "Spotify Linked";
             return Redirect(ManageSpotifyPath);
         }

Selector.Web/Extensions/ServiceExtensions.cs

@@ -9,10 +9,7 @@ namespace Selector.Web.Extensions
         public static void AddCacheHubProxy(this IServiceCollection services)
         {
             services.AddScoped<EventHubProxy>();
-            services.AddHostedService<CacheEventProxyService>();
+            services.AddHostedService<EventHubMappingService>();
-
-            services.AddTransient<ICacheEventMapping, NowPlayingCacheMapping>();
-            services.AddTransient<NowPlayingCacheMapping>();
 
             services.AddScoped<IEventHubMapping<NowPlayingHub, INowPlayingHubClient>, NowPlayingHubMapping>();
             services.AddScoped<NowPlayingHubMapping>();

Selector.Web/Selector.Web.csproj

@@ -10,6 +10,7 @@
     <ProjectReference Include="..\Selector\Selector.csproj" />
     <ProjectReference Include="..\Selector.Model\Selector.Model.csproj" />
     <ProjectReference Include="..\Selector.Cache\Selector.Cache.csproj" />
+    <ProjectReference Include="..\Selector.Event\Selector.Event.csproj" />
   </ItemGroup>
 
   <ItemGroup>

Selector.Web/Services/EventHubMappingService.cs

@@ -6,35 +6,27 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
+using Selector.Events;
+
 namespace Selector.Web.Service
 {
-    public class CacheEventProxyService: IHostedService
+    public class EventHubMappingService: IHostedService
     {
-        private readonly ILogger<CacheEventProxyService> Logger;
+        private readonly ILogger<EventHubMappingService> Logger;
         private readonly IServiceScopeFactory ScopeFactory;
 
-        private readonly IEnumerable<ICacheEventMapping> CacheEvents;
+        public EventHubMappingService(
-
+            ILogger<EventHubMappingService> logger,
-        public CacheEventProxyService(
-            ILogger<CacheEventProxyService> logger,
-            IEnumerable<ICacheEventMapping> mappings,
             IServiceScopeFactory scopeFactory
         )
         {
             Logger = logger;
             ScopeFactory = scopeFactory;
-
-            CacheEvents = mappings;
         }
 
         public Task StartAsync(CancellationToken cancellationToken)
         {
-            Logger.LogInformation("Starting cache event proxy");
+            Logger.LogInformation("Starting hub event mapping service");
-
-            foreach (var mapping in CacheEvents)
-            {
-                mapping.ConstructMapping();
-            }
 
             using (var scope = ScopeFactory.CreateScope())
             {
@@ -47,7 +39,6 @@ namespace Selector.Web.Service
 
         public Task StopAsync(CancellationToken cancellationToken)
         {
-            Logger.LogInformation("Stopping cache hub proxy");
             return Task.CompletedTask;
         }
     }

Selector.Web/Services/EventMappings/NowPlayingHubMapping.cs

@@ -3,7 +3,7 @@ using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Logging;
 
 using Selector.Web.Hubs;
-using Selector.Model.Events;
+using Selector.Events;
 
 namespace Selector.Web.Service
 {

Selector.Web/Startup.cs

@@ -11,6 +11,7 @@ using Microsoft.Extensions.Hosting;
 using Microsoft.AspNetCore.Identity;
 using Microsoft.EntityFrameworkCore;
 
+using Selector.Events;
 using Selector.Web.Hubs;
 using Selector.Web.Extensions;
 using Selector.Extensions;
@@ -93,20 +94,28 @@ namespace Selector.Web
             });
 
             services.AddAuthorisationHandlers();
-            services.AddModelEventBus();
 
-            if (config.RedisOptions.Enabled)
+            services.AddEvents();
-                services.AddRedisServices(config.RedisOptions.ConnectionString);
 
             services.AddSpotify();
+            ConfigureLastFm(config, services);
+
             if (config.RedisOptions.Enabled)
             {
+                Console.WriteLine("> Adding Redis...");
+                services.AddRedisServices(config.RedisOptions.ConnectionString);
+
+                Console.WriteLine("> Adding cache event maps...");
+
+                services.AddTransient<IEventMapping, SpotifyLinkToCacheMapping>();
+                services.AddTransient<IEventMapping, LastfmToCacheMapping>();
+                services.AddTransient<IEventMapping, NowPlayingFromCacheMapping>();
+
+                services.AddCacheHubProxy();
+
                 Console.WriteLine("> Adding caching Spotify consumers...");
                 services.AddCachingSpotify();
-                services.AddCacheHubProxy();
             }
-
-            ConfigureLastFm(config, services);
         }
 
         // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.

Selector.sln

@@ -1,7 +1,7 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 16
+# Visual Studio Version 17
-VisualStudioVersion = 16.0.30717.126
+VisualStudioVersion = 17.0.31919.166
 MinimumVisualStudioVersion = 15.0.26124.0
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Selector", "Selector\Selector.csproj", "{05B22ACE-2EA1-46AA-8483-A625B08A0D01}"
 EndProject
@@ -18,9 +18,11 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Selector.CLI", "Selector.CL
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Selector.Model", "Selector.Model\Selector.Model.csproj", "{B609975D-7CA6-422E-8461-E837C9EDB104}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Selector.Web", "Selector.Web\Selector.Web.csproj", "{ABC6EEBB-4C0D-45BD-8DDC-0B0304EAF34F}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Selector.Web", "Selector.Web\Selector.Web.csproj", "{ABC6EEBB-4C0D-45BD-8DDC-0B0304EAF34F}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Selector.Cache", "Selector.Cache\Selector.Cache.csproj", "{D8761D46-EF2B-4323-894F-E67C3EB0D0BB}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Selector.Cache", "Selector.Cache\Selector.Cache.csproj", "{D8761D46-EF2B-4323-894F-E67C3EB0D0BB}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Selector.Event", "Selector.Event\Selector.Event.csproj", "{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -52,6 +54,10 @@ Global
 		{D8761D46-EF2B-4323-894F-E67C3EB0D0BB}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{D8761D46-EF2B-4323-894F-E67C3EB0D0BB}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{D8761D46-EF2B-4323-894F-E67C3EB0D0BB}.Release|Any CPU.Build.0 = Release|Any CPU
+		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{C2FF1673-CB1A-43B7-A814-07BB3CB3A0D6}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

Selector/Consumers/PlayCounter.cs

@@ -8,6 +8,8 @@ using Microsoft.Extensions.Logging.Abstractions;
 
 using SpotifyAPI.Web;
 using IF.Lastfm.Core.Api;
+using IF.Lastfm.Core.Objects;
+using IF.Lastfm.Core.Api.Helpers;
 
 namespace Selector
 {
@@ -18,7 +20,7 @@ namespace Selector
         protected readonly IAlbumApi AlbumClient;
         protected readonly IArtistApi ArtistClient;
         protected readonly IUserApi UserClient;
-        protected readonly LastFmCredentials Credentials;
+        public readonly LastFmCredentials Credentials;
         protected readonly ILogger<PlayCounter> Logger;
 
         protected event EventHandler<PlayCount> NewPlayCount;
@@ -66,6 +68,12 @@ namespace Selector
 
         public async Task AsyncCallback(ListeningChangeEventArgs e)
         {
+            if(Credentials is null || string.IsNullOrWhiteSpace(Credentials.Username))
+            {
+                Logger.LogDebug("No Last.fm username, skipping play count");
+                return;
+            }
+
             if (e.Current.Item is FullTrack track)
             {
                 Logger.LogTrace("Making Last.fm call");
@@ -73,7 +81,6 @@ namespace Selector
                 var trackInfo = TrackClient.GetInfoAsync(track.Name, track.Artists[0].Name, username: Credentials?.Username);
                 var albumInfo = AlbumClient.GetInfoAsync(track.Album.Artists[0].Name, track.Album.Name, username: Credentials?.Username);
                 var artistInfo = ArtistClient.GetInfoAsync(track.Artists[0].Name);
-                // TODO: Null checking on credentials
                 var userInfo = UserClient.GetInfoAsync(Credentials.Username);
 
                 await Task.WhenAll(new Task[] { trackInfo, albumInfo, artistInfo, userInfo });

Selector/Watcher/Collection/Interfaces/IWatcherCollection.cs

@@ -11,13 +11,13 @@ namespace Selector
         /// Add watcher to collection, will start watcher if collection is running
         /// </summary>
         /// <param name="watcher">New watcher</param>
-        public void Add(IWatcher watcher);
+        public IWatcherContext Add(IWatcher watcher);
         /// <summary>
         /// Add watcher with given consumers to collection, will start watcher if collection is running
         /// </summary>
         /// <param name="watcher">New watcher</param>
         /// <param name="consumers">Consumers to subscribe to new watcher</param>
-        public void Add(IWatcher watcher, List<IConsumer> consumers);
+        public IWatcherContext Add(IWatcher watcher, List<IConsumer> consumers);
 
         /// <summary>
         /// Start watcher collection
@@ -27,5 +27,7 @@ namespace Selector
         /// Stop watcher collection
         /// </summary>
         public void Stop();
+
+        public IEnumerable<IConsumer> Consumers { get; }
     }
 }

Selector/Watcher/Collection/WatcherCollection.cs

@@ -22,6 +22,12 @@ namespace Selector
         }
 
         public int Count => Watchers.Count;
+
+        public IEnumerable<IConsumer> Consumers
+            => Watchers
+                .SelectMany(w => w.Consumers)
+                .Where(t => t is not null);
+
         public IEnumerable<Task> Tasks 
             => Watchers
                 .Select(w => w.Task)
@@ -32,17 +38,18 @@ namespace Selector
                 .Select(w => w.TokenSource)
                 .Where(t => t is not null);
 
-        public void Add(IWatcher watcher)
+        public IWatcherContext Add(IWatcher watcher)
         {
-            Add(watcher, default);
+            return Add(watcher, default);
         }
 
-        public void Add(IWatcher watcher, List<IConsumer> consumers)
+        public IWatcherContext Add(IWatcher watcher, List<IConsumer> consumers)
         {
             var context = WatcherContext.From(watcher, consumers);
             if (IsRunning) context.Start();
 
             Watchers.Add(context);
+            return context;
         }
 
         public IEnumerable<WatcherContext> Running
@@ -50,6 +57,8 @@ namespace Selector
 
         public void Start()
         {
+            if (IsRunning) return;
+
             Logger.LogDebug($"Starting {Count} watcher(s)");
             foreach(var watcher in Watchers)
             {
@@ -60,13 +69,33 @@ namespace Selector
         
         public void Stop()
         {
-            Logger.LogDebug($"Stopping {Count} watcher(s)");
+            if (!IsRunning) return;
-            foreach (var watcher in Watchers)
+
+            try
             {
-                watcher.Stop();
+                Logger.LogDebug($"Stopping {Count} watcher(s)");
+                foreach (var watcher in Watchers)
+                {
+                    watcher.Stop();
+                }
+                Task.WaitAll(Tasks.ToArray());
+                IsRunning = false;
+            }
+            catch (TaskCanceledException)
+            {
+                Logger.LogTrace("Caught task cancelled exception");
+            }
+            catch (AggregateException ex)
+            {
+                if(ex.InnerException is TaskCanceledException || ex.InnerExceptions.Any(e => e is TaskCanceledException))
+                {
+                    Logger.LogTrace("Caught task cancelled exception");
+                }
+                else
+                {
+                    throw ex;
+                }
             }
-            Task.WaitAll(Tasks.ToArray());
-            IsRunning = false;
         }
 
         public void Dispose()

Selector/Watcher/Context/WatcherContext.cs

@@ -10,7 +10,7 @@ namespace Selector
     public class WatcherContext: IDisposable, IWatcherContext
     {
         public IWatcher Watcher { get; set; }
-        private List<IConsumer> Consumers { get; set; } = new();
+        public List<IConsumer> Consumers { get; private set; } = new();
         public bool IsRunning { get; private set; }
         /// <summary>
         /// Reference to Watcher.Watch() task when running