feat: initialize manufacturing progress dashboard

2026-07-13 18:07:59 +08:00
commit 658a23b630
87 changed files with 19195 additions and 0 deletions
@@ -0,0 +1,5 @@
+using Microsoft.AspNetCore.SignalR;
+
+namespace DongfangHydro.Dashboard.Api.Realtime;
+
+public sealed class DashboardHub : Hub;
@@ -0,0 +1,112 @@
+using System.Collections.Concurrent;
+using DongfangHydro.Dashboard.Api.Contracts;
+using DongfangHydro.Dashboard.Api.Services;
+
+namespace DongfangHydro.Dashboard.Api.Realtime;
+
+public sealed class DashboardUpdateQueue(
+    IDashboardNotifier notifier,
+    IServiceScopeFactory scopeFactory,
+    ILogger<DashboardUpdateQueue> logger)
+    : BackgroundService, IDashboardUpdateQueue
+{
+    private const int MaximumAttempts = 5;
+    private readonly ConcurrentDictionary<Guid, QueuedUpdate> pendingByOrder = new();
+    private readonly SemaphoreSlim signal = new(0, 1);
+    private int signalScheduled;
+
+    public ValueTask EnqueueAsync(
+        UpdateNodeProgressResultDto update,
+        CancellationToken cancellationToken)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+        pendingByOrder.AddOrUpdate(
+            update.Order.Id,
+            _ => new QueuedUpdate(update),
+            (_, _) => new QueuedUpdate(update));
+        ScheduleWork();
+        return ValueTask.CompletedTask;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            await signal.WaitAsync(stoppingToken);
+            Interlocked.Exchange(ref signalScheduled, 0);
+
+            foreach (var orderId in pendingByOrder.Keys)
+            {
+                if (pendingByOrder.TryRemove(orderId, out var item))
+                {
+                    await PublishLatestAsync(orderId, item, stoppingToken);
+                }
+            }
+
+            if (!pendingByOrder.IsEmpty)
+            {
+                ScheduleWork();
+            }
+        }
+    }
+
+    private async Task PublishLatestAsync(
+        Guid orderId,
+        QueuedUpdate initial,
+        CancellationToken stoppingToken)
+    {
+        var current = initial;
+        var attempt = 1;
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            if (pendingByOrder.TryRemove(orderId, out var newer))
+            {
+                current = newer;
+                attempt = 1;
+            }
+
+            try
+            {
+                await using var scope = scopeFactory.CreateAsyncScope();
+                var overview = await scope.ServiceProvider
+                    .GetRequiredService<DashboardQueryService>()
+                    .GetOverviewAsync(stoppingToken);
+                await notifier.PublishUpdateAsync(current.Update, overview, stoppingToken);
+                return;
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                return;
+            }
+            catch (Exception exception)
+            {
+                logger.LogWarning(
+                    exception,
+                    "Dashboard update broadcast failed on attempt {Attempt} for order {OrderId}.",
+                    attempt,
+                    orderId);
+
+                if (attempt >= MaximumAttempts)
+                {
+                    return;
+                }
+
+                if (!pendingByOrder.ContainsKey(orderId))
+                {
+                    await Task.Delay(TimeSpan.FromSeconds(attempt * 2), stoppingToken);
+                    attempt += 1;
+                }
+            }
+        }
+    }
+
+    private void ScheduleWork()
+    {
+        if (Interlocked.CompareExchange(ref signalScheduled, 1, 0) == 0)
+        {
+            signal.Release();
+        }
+    }
+
+    private sealed record QueuedUpdate(UpdateNodeProgressResultDto Update);
+}
@@ -0,0 +1,11 @@
+using DongfangHydro.Dashboard.Api.Contracts;
+
+namespace DongfangHydro.Dashboard.Api.Realtime;
+
+public interface IDashboardNotifier
+{
+    Task PublishUpdateAsync(
+        UpdateNodeProgressResultDto update,
+        DashboardOverviewDto overview,
+        CancellationToken cancellationToken);
+}
@@ -0,0 +1,10 @@
+using DongfangHydro.Dashboard.Api.Contracts;
+
+namespace DongfangHydro.Dashboard.Api.Realtime;
+
+public interface IDashboardUpdateQueue
+{
+    ValueTask EnqueueAsync(
+        UpdateNodeProgressResultDto update,
+        CancellationToken cancellationToken);
+}
@@ -0,0 +1,24 @@
+using DongfangHydro.Dashboard.Api.Contracts;
+using Microsoft.AspNetCore.SignalR;
+
+namespace DongfangHydro.Dashboard.Api.Realtime;
+
+public sealed class SignalRDashboardNotifier(IHubContext<DashboardHub> hubContext)
+    : IDashboardNotifier
+{
+    public async Task PublishUpdateAsync(
+        UpdateNodeProgressResultDto update,
+        DashboardOverviewDto overview,
+        CancellationToken cancellationToken)
+    {
+        var clients = hubContext.Clients.All;
+        await Task.WhenAll(
+            clients.SendAsync("nodeUpdated", update.Node, cancellationToken),
+            clients.SendAsync("orderUpdated", update.Order, cancellationToken),
+            clients.SendAsync("trendAppended", update.TrendPoint, cancellationToken),
+            clients.SendAsync("overviewUpdated", overview, cancellationToken),
+            update.RiskEvent is null
+                ? Task.CompletedTask
+                : clients.SendAsync("riskEventCreated", update.RiskEvent, cancellationToken));
+    }
+}