fix: align backend APIs and upload flow

ConnectorService/Hubs/ChatHub.cs

@@ -10,11 +10,13 @@ namespace ConnectorService.Hubs
     {
         private readonly IConversationIntergrationService conService;
         private readonly StackExchange.Redis.IDatabase redis;
+        private readonly ConnectionRegistry registry;
 
-        public ChatHub(IConversationIntergrationService conService, IConnectionMultiplexer multiplexer)
+        public ChatHub(IConversationIntergrationService conService, IConnectionMultiplexer multiplexer, ConnectionRegistry registry)
         {
             this.conService = conService;
             this.redis = multiplexer.GetDatabase();
+            this.registry = registry;
         }
 
         public async override Task OnConnectedAsync()
@@ -34,6 +36,7 @@ namespace ConnectorService.Hubs
             }
 
             await redis.SetAddAsync(RedisHelper.GetConnectionIdKey(userId), Context.ConnectionId);
+            await registry.Add(Context);
 
 
             await base.OnConnectedAsync();
@@ -41,6 +44,7 @@ namespace ConnectorService.Hubs
 
         public async override Task OnDisconnectedAsync(Exception? exception)
         {
+            await registry.Remove(Context);
             if (Context.User.Identity.IsAuthenticated)
             {
                 var userId = Context.User.FindFirstValue(ClaimTypes.NameIdentifier);

ConnectorService/Program.cs

@@ -1,5 +1,7 @@
+using IM.InitCommon.Management;
 
 using ConnectorService.Hubs;
+using ConnectorService.Services;
 using IM.InitCommon;
 
 namespace ConnectorService
@@ -15,6 +17,8 @@ namespace ConnectorService
             builder.ConfigureDbConfiguration();
 
             builder.Services.AddSignalR();
+            builder.Services.AddSingleton<ConnectionRegistry>();
+            builder.Services.AddHostedService(sp => sp.GetRequiredService<ConnectionRegistry>());
 
             // Learn more about configuring Swagger/OpenAPI at https://aka.ms/aspnetcore/swashbuckle
             builder.Services.AddEndpointsApiExplorer();
@@ -23,6 +27,7 @@ namespace ConnectorService
             builder.ConfigExtraServices();
 
             var app = builder.Build();
+            if (app.ApplyMigrationsIfRequested(args)) return;
 
             // Configure the HTTP request pipeline.
             if (app.Environment.IsDevelopment())
@@ -32,9 +37,12 @@ namespace ConnectorService
             }
 
             app.UseAppDefault();
+            app.MapManagementHealth();
 
 
             app.MapHub<ChatHub>("/chat");
+            app.MapGet("/internal/management/connections", async (ConnectionRegistry registry) => await registry.Counts());
+            app.MapPost("/internal/management/disconnect/{id:guid}", async (Guid id, ConnectionRegistry registry) => { await registry.Disconnect(id); return new { disconnected = true }; });
 
             app.Run();
         }

ConnectorService/Services/ConnectionRegistry.cs

@@ -0,0 +1,38 @@
+using System.Collections.Concurrent;
+using System.Security.Claims;
+using Microsoft.AspNetCore.SignalR;
+using StackExchange.Redis;
+
+namespace ConnectorService.Services;
+public sealed class ConnectionRegistry(IConnectionMultiplexer redis, ILogger<ConnectionRegistry> logger) : BackgroundService
+{
+    const string Active = "im:management:connections";
+    static readonly RedisChannel Revocations = RedisChannel.Literal("im:management:disconnect");
+    readonly ConcurrentDictionary<string, HubCallerContext> contexts = new();
+    public async Task Add(HubCallerContext c) { contexts[c.ConnectionId] = c; await Beat(c); }
+    public async Task Remove(HubCallerContext c) { contexts.TryRemove(c.ConnectionId, out _); await redis.GetDatabase().SortedSetRemoveAsync(Active, Key(c)); }
+    static string Key(HubCallerContext c) => c.User?.FindFirstValue(ClaimTypes.NameIdentifier) + ":" + c.ConnectionId;
+    Task Beat(HubCallerContext c) => redis.GetDatabase().SortedSetAddAsync(Active, Key(c), DateTimeOffset.UtcNow.AddSeconds(90).ToUnixTimeSeconds());
+    public async Task Disconnect(Guid id) { Abort(id.ToString()); await redis.GetSubscriber().PublishAsync(Revocations, id.ToString()); }
+    void Abort(string userId) { foreach (var c in contexts.Values.Where(c => c.User!.FindFirstValue(ClaimTypes.NameIdentifier) == userId)) c.Abort(); }
+    public async Task<object> Counts() {
+        var db = redis.GetDatabase(); await db.SortedSetRemoveRangeByScoreAsync(Active, double.NegativeInfinity, DateTimeOffset.UtcNow.ToUnixTimeSeconds());
+        var active = await db.SortedSetRangeByScoreAsync(Active, DateTimeOffset.UtcNow.ToUnixTimeSeconds(), double.PositiveInfinity);
+        return new { connections = active.Length, users = active.Select(x => x.ToString().Split(':')[0]).Distinct().Count(), sampledAt = DateTime.UtcNow, expiresAfterSeconds = 90 };
+    }
+    protected override async Task ExecuteAsync(CancellationToken ct) {
+        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(20));
+        var subscribed = false;
+        try { do {
+            try {
+                if (!subscribed) { await redis.GetSubscriber().SubscribeAsync(Revocations, (_, value) => Abort(value.ToString())); subscribed = true; }
+                foreach (var c in contexts.Values) { if (c.ConnectionAborted.IsCancellationRequested) await Remove(c); else await Beat(c); }
+            } catch (Exception e) when (!ct.IsCancellationRequested) {
+                // Fail closed while the revocation channel is unavailable.
+                foreach (var c in contexts.Values) c.Abort();
+                logger.LogWarning("Connection heartbeat unavailable: {Type}",e.GetType().Name);
+            }
+        } while (await timer.WaitForNextTickAsync(ct)); }
+        finally { try { await redis.GetSubscriber().UnsubscribeAsync(Revocations); } catch { } }
+    }
+}