feat: board-first orchestration with Gateway Bridge, live-update, and flow-board

- GatewayBridgeController: MCP-artiger Kommando-Adapter für Agent-zu-Backend - TaskBridgeService + LiveUpdateService: SSE Live-Sync + Bridge-Kommandos - FlowBoard.vue: Board-first orchestration dashboard panel - live-sync.ts store + live.ts service: SSE-basierte Live-Updates - Nullability-Warnung in HealthController.cs gefixt - nginx.conf: SSE-Proxy + CORS für Bridge-Endpunkte - .gitignore: pnpm/corepack local caches ausgeschlossen - docs: architecture-board-first-orchestration.md hinzugefügt - README: Backend Bridge API dokumentiert
2026-06-22 19:56:45 +02:00
parent de1fc198cb
commit df94ed3cd4
34 changed files with 2115 additions and 118 deletions
@@ -0,0 +1,87 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
+using Nexus.Api.Models;
+
+namespace Nexus.Api.Services;
+
+public sealed class LiveUpdateService : ILiveUpdateService
+{
+    private const int ReplayLimit = 256;
+
+    private readonly ConcurrentDictionary<Guid, Channel<LiveUpdateEnvelope>> _subscribers = new();
+    private readonly object _historyLock = new();
+    private readonly Queue<LiveUpdateEnvelope> _history = new();
+    private long _sequence;
+
+    public long CurrentSequence => Interlocked.Read(ref _sequence);
+
+    public Task<LiveUpdateSubscription> SubscribeAsync(long? afterSequence = null, CancellationToken ct = default)
+    {
+        var channel = Channel.CreateUnbounded<LiveUpdateEnvelope>(new UnboundedChannelOptions
+        {
+            SingleReader = true,
+            SingleWriter = false,
+            AllowSynchronousContinuations = false
+        });
+
+        var id = Guid.NewGuid();
+        _subscribers[id] = channel;
+
+        var replay = afterSequence.HasValue ? GetReplay(afterSequence.Value) : Array.Empty<LiveUpdateEnvelope>();
+        foreach (var envelope in replay)
+        {
+            channel.Writer.TryWrite(envelope);
+        }
+
+        ct.Register(() =>
+        {
+            if (_subscribers.TryRemove(id, out var removed))
+            {
+                removed.Writer.TryComplete();
+            }
+        });
+
+        return Task.FromResult(new LiveUpdateSubscription
+        {
+            Reader = channel.Reader,
+            StartingSequence = replay.LastOrDefault()?.Sequence ?? CurrentSequence
+        });
+    }
+
+    public LiveUpdateEnvelope Publish(string type, object payload, string channel = "dashboard")
+    {
+        var envelope = new LiveUpdateEnvelope(
+            type,
+            DateTimeOffset.UtcNow,
+            payload,
+            Interlocked.Increment(ref _sequence),
+            channel);
+
+        lock (_historyLock)
+        {
+            _history.Enqueue(envelope);
+            while (_history.Count > ReplayLimit)
+            {
+                _history.Dequeue();
+            }
+        }
+
+        foreach (var (id, subscriber) in _subscribers)
+        {
+            if (!subscriber.Writer.TryWrite(envelope) && _subscribers.TryRemove(id, out var removed))
+            {
+                removed.Writer.TryComplete();
+            }
+        }
+
+        return envelope;
+    }
+
+    private LiveUpdateEnvelope[] GetReplay(long afterSequence)
+    {
+        lock (_historyLock)
+        {
+            return _history.Where(item => item.Sequence > afterSequence).ToArray();
+        }
+    }
+}