From 990a528d676f121d51c228b6ea6438284d52fe98 Mon Sep 17 00:00:00 2001 From: Vitali sharp8n Date: Mon, 7 Sep 2026 22:05:15 +0300 Subject: [PATCH] workflows --- Controllers/RealtimeController.cs | 87 ++++++++++++++++++++++++++ Controllers/WorkflowFilesController.cs | 6 ++ Program.cs | 3 + Services/RealtimeEventHub.cs | 47 ++++++++++++++ 4 files changed, 143 insertions(+) create mode 100644 Controllers/RealtimeController.cs create mode 100644 Services/RealtimeEventHub.cs diff --git a/Controllers/RealtimeController.cs b/Controllers/RealtimeController.cs new file mode 100644 index 0000000..5f30b27 --- /dev/null +++ b/Controllers/RealtimeController.cs @@ -0,0 +1,87 @@ +using System.Text.Json; +using Microsoft.AspNetCore.Mvc; +using w4c_workflows.Services; + +namespace w4c_workflows.Controllers; + +/// +/// Tenant-scoped server-sent-events stream: GET /api/workflows/events pushes +/// data: {"kind":"workflow","id":"","at":""} frames whenever a workflow file is +/// written server-side. The client re-fetches the touched file; no payload is pushed. +/// The full literal route avoids colliding with api/workflows/{id} and routes to this service +/// through the SPA proxy (/api/workflows prefix). +/// +[ApiController] +[Route("api/workflows/events")] +public class RealtimeController : ControllerBase +{ + private static readonly JsonSerializerOptions JsonOptions = new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase }; + private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(15); + + private readonly RealtimeEventHub _hub; + private readonly ILogger _logger; + + public RealtimeController(RealtimeEventHub hub, ILogger logger) + { + _hub = hub; + _logger = logger; + } + + private string TenantId => + (string?)HttpContext.Items["TenantId"] + ?? throw new InvalidOperationException("TenantId not resolved by auth middleware"); + + [HttpGet] + public async Task Stream(CancellationToken ct) + { + var tid = TenantId; + + Response.Headers["Content-Type"] = "text/event-stream"; + Response.Headers["Cache-Control"] = "no-cache"; + Response.Headers["Connection"] = "keep-alive"; + Response.Headers["X-Accel-Buffering"] = "no"; + await Response.Body.FlushAsync(ct); + + var (id, reader) = _hub.AddClient(tid); + try + { + while (!ct.IsCancellationRequested) + { + using var readCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + readCts.CancelAfter(HeartbeatInterval); + bool hasItem; + try + { + hasItem = await reader.WaitToReadAsync(readCts.Token); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + await Response.WriteAsync(": ping\n\n", ct); + await Response.Body.FlushAsync(ct); + continue; + } + + if (!hasItem) break; + + while (reader.TryRead(out var evt)) + { + var json = JsonSerializer.Serialize(evt, JsonOptions); + await Response.WriteAsync($"data: {json}\n\n", ct); + } + await Response.Body.FlushAsync(ct); + } + } + catch (OperationCanceledException) + { + // Client disconnected — normal terminal state. + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Realtime SSE stream ended for tenant {Tenant}", tid); + } + finally + { + _hub.RemoveClient(tid, id); + } + } +} diff --git a/Controllers/WorkflowFilesController.cs b/Controllers/WorkflowFilesController.cs index 81be7e6..61e10eb 100644 --- a/Controllers/WorkflowFilesController.cs +++ b/Controllers/WorkflowFilesController.cs @@ -32,17 +32,20 @@ public class WorkflowFilesController : ControllerBase WorkflowSourceFactory sourceFactory, WorkflowRepoStore repoStore, WorkflowSyncService sync, + RealtimeEventHub events, ILogger logger) { _sourceFactory = sourceFactory; _repoStore = repoStore; _sync = sync; + _events = events; _logger = logger; } private readonly WorkflowSourceFactory _sourceFactory; private readonly WorkflowRepoStore _repoStore; private readonly WorkflowSyncService _sync; + private readonly RealtimeEventHub _events; private readonly ILogger _logger; private string TenantId => (string?)HttpContext.Items["TenantId"] @@ -225,6 +228,9 @@ public class WorkflowFilesController : ControllerBase System.IO.File.WriteAllText(full, request.Content ?? string.Empty, new System.Text.UTF8Encoding(false)); _logger.LogInformation("Saved workflow file {Path} for tenant {TenantId} ({Login})", request.Path, TenantId, MaybeForgejoLogin ?? "-"); + // Notify any open workflow editor that the file changed server-side (frontend is optional: + // the AI can write it via the backend while no page is open, then the page reloads on next visit). + _events.Publish(TenantId, "workflow", request.Path, DateTime.UtcNow); return Ok(new { success = true, path = request.Path }); } catch (Exception ex) diff --git a/Program.cs b/Program.cs index af4ecee..397910c 100644 --- a/Program.cs +++ b/Program.cs @@ -120,6 +120,9 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +// Realtime SSE hub — notifies subscribed clients when a workflow file changes server-side. +builder.Services.AddSingleton(); + // Workflow source: factory-based — each request creates a tenant-scoped source. // WorkflowSource:CopiesRoot is REQUIRED; the factory throws if it is empty, // preventing any request from serving cross-tenant workflows. diff --git a/Services/RealtimeEventHub.cs b/Services/RealtimeEventHub.cs new file mode 100644 index 0000000..c703d42 --- /dev/null +++ b/Services/RealtimeEventHub.cs @@ -0,0 +1,47 @@ +using System.Collections.Concurrent; +using System.Threading.Channels; + +namespace w4c_workflows.Services; + +/// +/// Server-push hub for tenant-scoped realtime events (SSE), mirroring w4c-webapi's hub. A connected +/// client gets its own unbounded channel; writes a tiny notification into every +/// channel for the tenant. The client decides what to re-fetch, keeping broadcasts cheap. +/// +/// Used to make the frontend an optional view: when a workflow file is written server-side (e.g. by +/// the AI transform), every subscribed client in the tenant is notified and reloads the file. +/// +public sealed class RealtimeEventHub +{ + private readonly ConcurrentDictionary>> _clients = new(); + + public (Guid Id, ChannelReader Reader) AddClient(string tenantId) + { + var channel = Channel.CreateUnbounded(); + var id = Guid.NewGuid(); + var clients = _clients.GetOrAdd(tenantId, _ => new ConcurrentDictionary>()); + clients[id] = channel; + return (id, channel.Reader); + } + + public void RemoveClient(string tenantId, Guid id) + { + if (!_clients.TryGetValue(tenantId, out var clients)) return; + clients.TryRemove(id, out _); + if (clients.IsEmpty) _clients.TryRemove(tenantId, out _); + } + + public void Publish(string tenantId, string kind, string id, DateTime at) + { + if (!_clients.TryGetValue(tenantId, out var clients)) return; + var evt = new RealtimeEvent(kind, id, at); + foreach (var channel in clients.Values) + { + channel.Writer.TryWrite(evt); + } + } +} + +/// A tenant-scoped change notification: Kind is a component name (e.g. "workflow"), +/// Id is that component's server id/path, At is the server-side write time. +public record RealtimeEvent(string Kind, string Id, DateTime At);