Notify users of who is spectating them

Server-side part of ppy/osu#22087.

SampleSpectatorClient/SpectatorClient.cs

@@ -64,6 +64,19 @@ Task ISpectatorClient.UserScoreProcessed(int userId, long scoreId)
             return Task.CompletedTask;
         }
 
+        Task ISpectatorClient.UserStartedWatching(SpectatorUser[] users)
+        {
+            foreach (var user in users)
+                Console.WriteLine($"{connection.ConnectionId} User {user.OnlineID} started watching you");
+            return Task.CompletedTask;
+        }
+
+        Task ISpectatorClient.UserEndedWatching(int userId)
+        {
+            Console.WriteLine($"{connection.ConnectionId} User {userId} ended watching you");
+            return Task.CompletedTask;
+        }
+
         public Task BeginPlaying(long? scoreToken, SpectatorState state) => connection.SendAsync(nameof(ISpectatorServer.BeginPlaySession), scoreToken, state);
 
         public Task SendFrames(FrameDataBundle data) => connection.SendAsync(nameof(ISpectatorServer.SendFrameData), data);

osu.Server.Spectator.Tests/SpectatorHubTest.cs

@@ -10,7 +10,6 @@
 using Moq;
 using osu.Game.Beatmaps;
 using osu.Game.Online.API;
-using osu.Game.Online.API.Requests.Responses;
 using osu.Game.Online.Spectator;
 using osu.Game.Replays.Legacy;
 using osu.Game.Rulesets.Osu.Mods;
@@ -42,6 +41,7 @@ public class SpectatorHubTest
         public SpectatorHubTest()
         {
             var clientStates = new EntityStore<SpectatorClientState>();
+            var spectatorLists = new EntityStore<SpectatorList>();
 
             mockDatabase = new Mock<IDatabaseAccess>();
             mockDatabase.Setup(db => db.GetUsernameAsync(streamer_id)).ReturnsAsync(() => streamer_username);
@@ -65,7 +65,7 @@ public SpectatorHubTest()
 
             var mockScoreProcessedSubscriber = new Mock<IScoreProcessedSubscriber>();
 
-            hub = new SpectatorHub(loggerFactory.Object, clientStates, databaseFactory.Object, scoreUploader, mockScoreProcessedSubscriber.Object);
+            hub = new SpectatorHub(loggerFactory.Object, clientStates, spectatorLists, databaseFactory.Object, scoreUploader, mockScoreProcessedSubscriber.Object);
         }
 
         [Fact]
@@ -325,9 +325,12 @@ public async Task NewUserBeginsWatchingStream(bool ongoing)
 
             Mock<IHubCallerClients<ISpectatorClient>> mockClients = new Mock<IHubCallerClients<ISpectatorClient>>();
             Mock<ISpectatorClient> mockCaller = new Mock<ISpectatorClient>();
+            Mock<ISpectatorClient> mockStreamer = new Mock<ISpectatorClient>();
 
             mockClients.Setup(clients => clients.Caller).Returns(mockCaller.Object);
             mockClients.Setup(clients => clients.All).Returns(mockCaller.Object);
+            mockClients.Setup(clients => clients.User(streamer_id.ToString())).Returns(mockStreamer.Object);
+            mockDatabase.Setup(db => db.GetUsernameAsync(watcher_id)).ReturnsAsync("watcher");
 
             Mock<IGroupManager> mockGroups = new Mock<IGroupManager>();
 
@@ -362,6 +365,12 @@ public async Task NewUserBeginsWatchingStream(bool ongoing)
             mockGroups.Verify(groups => groups.AddToGroupAsync(connectionId, SpectatorHub.GetGroupId(streamer_id), default));
 
             mockCaller.Verify(clients => clients.UserBeganPlaying(streamer_id, It.Is<SpectatorState>(m => m.Equals(state))), Times.Exactly(ongoing ? 2 : 0));
+            mockStreamer.Verify(client => client.UserStartedWatching(It.Is<SpectatorUser[]>(users => users.Single().OnlineID == watcher_id)), Times.Once);
+
+            await hub.EndWatchingUser(streamer_id);
+
+            mockGroups.Verify(groups => groups.RemoveFromGroupAsync(connectionId, SpectatorHub.GetGroupId(streamer_id), default));
+            mockStreamer.Verify(client => client.UserEndedWatching(watcher_id), Times.Once);
         }
 
         [Fact]

osu.Server.Spectator/Extensions/ServiceCollectionExtensions.cs

@@ -18,6 +18,7 @@ public static class ServiceCollectionExtensions
         public static IServiceCollection AddHubEntities(this IServiceCollection serviceCollection)
         {
             return serviceCollection.AddSingleton<EntityStore<SpectatorClientState>>()
+                                    .AddSingleton<EntityStore<SpectatorList>>()
                                     .AddSingleton<EntityStore<MultiplayerClientState>>()
                                     .AddSingleton<EntityStore<ServerMultiplayerRoom>>()
                                     .AddSingleton<EntityStore<ConnectionState>>()

osu.Server.Spectator/Hubs/Spectator/SpectatorHub.cs

@@ -31,18 +31,21 @@ public class SpectatorHub : StatefulUserHub<ISpectatorClient, SpectatorClientSta
         /// </summary>
         private const BeatmapOnlineStatus max_beatmap_status_for_replays = BeatmapOnlineStatus.Loved;
 
+        private readonly EntityStore<SpectatorList> spectatorLists;
         private readonly IDatabaseFactory databaseFactory;
         private readonly ScoreUploader scoreUploader;
         private readonly IScoreProcessedSubscriber scoreProcessedSubscriber;
 
         public SpectatorHub(
             ILoggerFactory loggerFactory,
             EntityStore<SpectatorClientState> users,
+            EntityStore<SpectatorList> spectatorLists,
             IDatabaseFactory databaseFactory,
             ScoreUploader scoreUploader,
             IScoreProcessedSubscriber scoreProcessedSubscriber)
             : base(loggerFactory, users)
         {
+            this.spectatorLists = spectatorLists;
             this.databaseFactory = databaseFactory;
             this.scoreUploader = scoreUploader;
             this.scoreProcessedSubscriber = scoreProcessedSubscriber;
@@ -203,11 +206,47 @@ public async Task StartWatchingUser(int userId)
             }
 
             await Groups.AddToGroupAsync(Context.ConnectionId, GetGroupId(userId));
+
+            int watcherId = Context.GetUserId();
+            string? watcherUsername;
+            using (var db = databaseFactory.GetInstance())
+                watcherUsername = await db.GetUsernameAsync(watcherId);
+
+            if (watcherUsername == null)
+                return;
+
+            var watcher = new SpectatorUser
+            {
+                OnlineID = watcherId,
+                Username = watcherUsername,
+            };
+
+            using (var usage = await spectatorLists.GetForUse(userId, createOnMissing: true))
+            {
+                usage.Item ??= new SpectatorList();
+                usage.Item.Spectators[watcherId] = watcher;
+            }
+
+            await Clients.User(userId.ToString()).UserStartedWatching([watcher]);
         }
 
         public async Task EndWatchingUser(int userId)
         {
             await Groups.RemoveFromGroupAsync(Context.ConnectionId, GetGroupId(userId));
+
+            int watcherId = Context.GetUserId();
+
+            using (var usage = await spectatorLists.TryGetForUse(userId))
+            {
+                if (usage?.Item == null)
+                    return;
+
+                usage.Item.Spectators.Remove(watcherId);
+                if (usage.Item.Spectators.Count == 0)
+                    usage.Destroy();
+            }
+
+            await Clients.User(userId.ToString()).UserEndedWatching(watcherId);
         }
 
         public override async Task OnConnectedAsync()
@@ -217,6 +256,17 @@ public override async Task OnConnectedAsync()
             foreach (var kvp in GetAllStates())
                 await Clients.Caller.UserBeganPlaying((int)kvp.Key, kvp.Value.State!);
 
+            SpectatorUser[]? watchers = null;
+
+            using (var usage = await spectatorLists.TryGetForUse(Context.GetUserId()))
+            {
+                if (usage?.Item != null)
+                    watchers = usage.Item.Spectators.Values.ToArray();
+            }
+
+            if (watchers != null)
+                await Clients.Caller.UserStartedWatching(watchers);
+
             await base.OnConnectedAsync();
         }
 

osu.Server.Spectator/Hubs/Spectator/SpectatorList.cs

@@ -0,0 +1,13 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System.Collections.Generic;
+using osu.Game.Online.Spectator;
+
+namespace osu.Server.Spectator.Hubs.Spectator
+{
+    public class SpectatorList
+    {
+        public Dictionary<int, SpectatorUser> Spectators { get; } = new Dictionary<int, SpectatorUser>();
+    }
+}