feat: probe host stats, node cards with a separate lines panel grouped by carrier, Windows host sampling, manual carrier override [CI]

backend/src/MoShiro.Backend.Api/Public/LiveEndpoints.cs

@@ -32,7 +32,7 @@ public static class LiveEndpoints
     }
 
     private static async Task<IResult> StreamAsync(
-        HttpContext http, AgentService agents, LineService lineService, TestService tests, QueryService query, LiveHub hub, SseLimiter limiter,
+        HttpContext http, AgentService agents, ProbeService probes, HostStatusService hosts, LineService lineService, TestService tests, QueryService query, LiveHub hub, SseLimiter limiter,
         BackendOptions options, BackendCounters counters, TimeProvider time, IHostApplicationLifetime lifetime, string? ip_version)
     {
         string ipVersion = ip_version ?? IpVersions.V4;
@@ -84,7 +84,7 @@ public static class LiveEndpoints
             http.Response.Headers["X-Accel-Buffering"] = "no";
             await http.Response.Body.FlushAsync(http.RequestAborted);
 
-            var session = new LiveSession(http, agents, lineService, tests, query, hub, options, time, topics, lineKeys, ipVersion);
+            var session = new LiveSession(http, agents, probes, hosts, lineService, tests, query, hub, options, time, topics, lineKeys, ipVersion);
             using var linked = CancellationTokenSource.CreateLinkedTokenSource(http.RequestAborted, lifetime.ApplicationStopping);
             await session.RunAsync(linked.Token);
         }
@@ -105,6 +105,8 @@ public static class LiveEndpoints
     {
         private readonly HttpContext _http;
         private readonly AgentService _agents;
+        private readonly ProbeService _probes;
+        private readonly HostStatusService _hosts;
         private readonly LineService _lines;
         private readonly IReadOnlyDictionary<string, LineKey> _lineKeys;
         private readonly string _ipVersion;
@@ -124,10 +126,12 @@ public static class LiveEndpoints
         private long _lastWriteMs;
         private long _lastPeriodicMs;
 
-        public LiveSession(HttpContext http, AgentService agents, LineService lines, TestService tests, QueryService query, LiveHub hub, BackendOptions options, TimeProvider time, IReadOnlyList<LiveTopic> topics, IReadOnlyDictionary<string, LineKey> lineKeys, string ipVersion)
+        public LiveSession(HttpContext http, AgentService agents, ProbeService probes, HostStatusService hosts, LineService lines, TestService tests, QueryService query, LiveHub hub, BackendOptions options, TimeProvider time, IReadOnlyList<LiveTopic> topics, IReadOnlyDictionary<string, LineKey> lineKeys, string ipVersion)
         {
             _http = http;
             _agents = agents;
+            _probes = probes;
+            _hosts = hosts;
             _lines = lines;
             _lineKeys = lineKeys;
             _ipVersion = ipVersion;
@@ -171,7 +175,7 @@ public static class LiveEndpoints
                                 _nodeSignals.Enqueue(node);
                                 break;
                             case HostSignal host when _topics.Any(t => t.Kind == LiveTopicKind.Hosts):
-                                _hostSignals[host.AgentId] = host;
+                                _hostSignals[host.Node] = host;
                                 break;
                             case ChunkSignal chunk:
                                 _dirtyLines.Add((chunk.AgentId, chunk.Series));
@@ -216,10 +220,20 @@ public static class LiveEndpoints
 
             foreach (HostSignal host in hostSignals)
             {
-                AgentView? view = await _agents.GetAsync(host.AgentId, ct);
+                if (host.Node.Value.StartsWith(ProbeId.Prefix, StringComparison.Ordinal))
+                {
+                    if (await _probes.GetAsync(host.Node.Value, ct) is { Revoked: false } probe)
+                    {
+                        await WriteAsync("host", new LiveHostDto(host.Node.Value, "probe", probe.Online, Mappers.ToDto(host.Host)), ApiJsonContext.Default.LiveHostDto, ct);
+                    }
+
+                    continue;
+                }
+
+                AgentView? view = await _agents.GetAsync(host.Node, ct);
                 if (view is not null && IsVisible(view) && view.Agent.Status == Fleet.Server.NodeStatus.Active)
                 {
-                    await WriteAsync("host", new LiveHostDto(host.AgentId.Value, view.Online, Mappers.ToDto(host.Host)), ApiJsonContext.Default.LiveHostDto, ct);
+                    await WriteAsync("host", new LiveHostDto(host.Node.Value, "agent", view.Online, Mappers.ToDto(host.Host)), ApiJsonContext.Default.LiveHostDto, ct);
                 }
             }
 
@@ -284,7 +298,15 @@ public static class LiveEndpoints
                     case LiveTopicKind.Hosts:
                         foreach (AgentView view in (await _agents.ListAsync(ct)).Where(v => v.Agent.Status == Fleet.Server.NodeStatus.Active && IsVisible(v) && v.Host is not null))
                         {
-                            await WriteAsync("host", new LiveHostDto(view.Agent.Id.Value, view.Online, Mappers.ToDto(view.Host!)), ApiJsonContext.Default.LiveHostDto, ct);
+                            await WriteAsync("host", new LiveHostDto(view.Agent.Id.Value, "agent", view.Online, Mappers.ToDto(view.Host!)), ApiJsonContext.Default.LiveHostDto, ct);
+                        }
+
+                        foreach (ProbeView view in await _probes.ListAsync(ct))
+                        {
+                            if (!view.Revoked && _hosts.Latest(new NodeId(view.Probe.Id)).Host is { } host)
+                            {
+                                await WriteAsync("host", new LiveHostDto(view.Probe.Id, "probe", view.Online, Mappers.ToDto(host)), ApiJsonContext.Default.LiveHostDto, ct);
+                            }
                         }
 
                         break;