From bd41609424619544e02839215bed49bbc1d39500 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Wed, 2 Sep 2026 09:49:15 +0000 Subject: [PATCH] Add NATS JetStream publish + trigger activities NatsPublish (CodeActivity) publishes a UTF-8 body to a subject with optional Nats-Msg-Id dedup and ExpectedStream assertion. NatsTrigger (Trigger) starts a workflow per message on a specified stream/subject. A single BackgroundService scans the trigger store for indexed NatsTriggers, groups by (stream, subject), and runs one durable pull consumer per pair with Explicit ack, so multiple Elsa instances share the workload at-least-once. A mediator handler on WorkflowTriggersIndexed refreshes subscriptions when workflows are published. Wired via Nats:Url/User/Password (env-var overrides Nats__*, or short NATS_URL/NATS_USER/NATS_PASSWORD fallbacks). Uses the NATS.Net v2 client so wss:// works transparently. --- CLAUDE.md | 35 +++ .../Activities/NatsPublish.cs | 61 ++++++ .../Activities/NatsTrigger.cs | 72 +++++++ .../Elsa.CustomServer.csproj | 3 + .../Nats/NatsTriggerHostedService.cs | 200 ++++++++++++++++++ .../Nats/NatsTriggerIndexChangedHandler.cs | 13 ++ .../Nats/NatsTriggerStimulus.cs | 6 + src/Elsa.CustomServer/Program.cs | 32 +++ src/Elsa.CustomServer/appsettings.json | 5 + 9 files changed, 427 insertions(+) create mode 100644 src/Elsa.CustomServer/Activities/NatsPublish.cs create mode 100644 src/Elsa.CustomServer/Activities/NatsTrigger.cs create mode 100644 src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs create mode 100644 src/Elsa.CustomServer/Nats/NatsTriggerIndexChangedHandler.cs create mode 100644 src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs diff --git a/CLAUDE.md b/CLAUDE.md index 7af71e4..2b0ce4f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -169,6 +169,41 @@ to its DSN. at create time. `POST /applications/dockercompose` rejects it otherwise. - Deploys are fire-and-forget — never poll from Claude. User's global rule. +## NATS activities + +Three files under `src/Elsa.CustomServer/`: + +- `Activities/NatsPublish.cs` — publisher `CodeActivity`. Inputs: `Subject`, + `Body`, optional `MessageId` (sets the `Nats-Msg-Id` header for JetStream + server-side dedup), optional `ExpectedStream` (fail-fast assertion). Outputs + `Sequence` and `Duplicate`. +- `Activities/NatsTrigger.cs` — trigger `Trigger`. Inputs: `Stream`, + `Subject` (wildcards allowed). Outputs: `Result` = message body (UTF-8), + `ReceivedSubject`, `Headers`. Its trigger name / stimulus name is + `NATS.NatsTrigger`. +- `Nats/NatsTriggerHostedService.cs` — one `BackgroundService` that scans + `ITriggerStore` for stored triggers named `NATS.NatsTrigger`, spins up one + durable pull consumer per unique `(stream, subject)` pair, and dispatches + workflows via `IWorkflowDispatcher.DispatchAsync(DispatchTriggerWorkflowsRequest)` + as messages arrive. The mediator handler + `NatsTriggerIndexChangedHandler` re-runs the scan whenever + `WorkflowTriggersIndexed` fires (i.e. after publishing a workflow). + +Consumer model: shared durable pull consumer with `DurableName = +elsa--` and `AckPolicy.Explicit`. That means +if multiple Elsa instances are running, exactly one processes each message +(load-balanced), at-least-once. `MaxDeliver = 5`, `AckWait = 30s`, +`MaxAckPending = 256`. Failed dispatches NAK with a 5-second delay. + +Configuration: `Nats:Url`, `Nats:User`, `Nats:Password` in appsettings +(env-var overrides `Nats__Url`, `Nats__User`, `Nats__Password`, or the +short `NATS_URL` / `NATS_USER` / `NATS_PASSWORD` fallbacks). Leave `Nats:Url` +empty to disable NATS entirely — the activities still register in the toolbox +but publishing throws and no consumers are started. + +Packages: `NATS.Net` + `NATS.Client.Hosting` (v2 client). The v1 `NATS.Client` +package does not support `wss://` from .NET — do not use it. + ## Session-specific reminders - Custom activities added under `Activities/` in the SERVER project (not the diff --git a/src/Elsa.CustomServer/Activities/NatsPublish.cs b/src/Elsa.CustomServer/Activities/NatsPublish.cs new file mode 100644 index 0000000..509af1c --- /dev/null +++ b/src/Elsa.CustomServer/Activities/NatsPublish.cs @@ -0,0 +1,61 @@ +using System.Text; +using Elsa.Extensions; +using Elsa.Workflows; +using Elsa.Workflows.Attributes; +using Elsa.Workflows.Models; +using NATS.Client.Core; +using NATS.Client.JetStream; +using NATS.Client.JetStream.Models; + +namespace Elsa.CustomServer.Activities; + +[Activity("NATS", "NATS", "Publish a message to a NATS JetStream subject.", DisplayName = "NATS Publish", Kind = ActivityKind.Task)] +public class NatsPublish : CodeActivity +{ + [Input(Description = "Subject to publish to (must be bound to a JetStream stream).")] + public Input Subject { get; set; } = default!; + + [Input(Description = "Message body (UTF-8 string).")] + public Input Body { get; set; } = default!; + + [Input(Description = "Optional Nats-Msg-Id used by JetStream for server-side deduplication within the stream's duplicate window.")] + public Input MessageId { get; set; } = new(default(string?)); + + [Input(Description = "Optional stream name for a fail-fast assertion that the subject lands in this stream.")] + public Input ExpectedStream { get; set; } = new(default(string?)); + + [Output(Description = "Stream sequence number the server assigned to the accepted message.")] + public Output Sequence { get; set; } = default!; + + [Output(Description = "True when JetStream deduplicated (Nats-Msg-Id already seen within the duplicate window).")] + public Output Duplicate { get; set; } = default!; + + protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) + { + var subject = context.Get(Subject); + if (string.IsNullOrWhiteSpace(subject)) + throw new InvalidOperationException("NatsPublish: Subject is required."); + + var body = context.Get(Body) ?? string.Empty; + var messageId = context.Get(MessageId); + var expectedStream = context.Get(ExpectedStream); + + var nats = context.GetRequiredService(); + var js = new NatsJSContext((NatsConnection)nats); + + NatsHeaders? headers = null; + if (!string.IsNullOrEmpty(messageId)) + headers = new NatsHeaders { ["Nats-Msg-Id"] = messageId }; + + var opts = string.IsNullOrEmpty(expectedStream) + ? null + : new NatsJSPubOpts { ExpectedStream = expectedStream }; + + var payload = Encoding.UTF8.GetBytes(body); + var ack = await js.PublishAsync(subject, payload, headers: headers, opts: opts, cancellationToken: context.CancellationToken); + ack.EnsureSuccess(); + + context.Set(Sequence, (long)ack.Seq); + context.Set(Duplicate, ack.Duplicate); + } +} diff --git a/src/Elsa.CustomServer/Activities/NatsTrigger.cs b/src/Elsa.CustomServer/Activities/NatsTrigger.cs new file mode 100644 index 0000000..34f3b5e --- /dev/null +++ b/src/Elsa.CustomServer/Activities/NatsTrigger.cs @@ -0,0 +1,72 @@ +using Elsa.CustomServer.Nats; +using Elsa.Extensions; +using Elsa.Workflows; +using Elsa.Workflows.Attributes; +using Elsa.Workflows.Models; + +namespace Elsa.CustomServer.Activities; + +[Activity("NATS", "NATS", "Start a workflow when a message arrives on a NATS JetStream subject.", DisplayName = "NATS Trigger", Kind = ActivityKind.Trigger)] +public class NatsTrigger : Trigger +{ + // Name that identifies stored triggers of this type in the trigger store — + // the hosted subscriber uses it to find which streams/subjects to consume. + public const string StimulusName = "NATS.NatsTrigger"; + + // Workflow-input keys the hosted service passes when it dispatches a new instance. + public const string BodyInputKey = "NatsBody"; + public const string SubjectInputKey = "NatsSubject"; + public const string HeadersInputKey = "NatsHeaders"; + + [Input(Description = "JetStream name that owns the subject.")] + public Input Stream { get; set; } = default!; + + [Input(Description = "Subject to subscribe to. Wildcards are allowed: 'orders.*' matches one token, 'orders.>' matches one or more.")] + public Input Subject { get; set; } = default!; + + [Output(Description = "The concrete subject the message was published to (useful with wildcards).")] + public Output ReceivedSubject { get; set; } = default!; + + [Output(Description = "Message headers as a string dictionary.")] + public Output> Headers { get; set; } = default!; + + protected override IEnumerable GetTriggerPayloads(TriggerIndexingContext context) + { + context.TriggerName = StimulusName; + var stream = Stream.Get(context.ExpressionExecutionContext); + var subject = Subject.Get(context.ExpressionExecutionContext); + return [new NatsTriggerStimulus(stream, subject)]; + } + + protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) + { + if (context.IsTriggerOfWorkflow()) + { + EmitOutputs(context); + await context.CompleteActivityAsync(); + return; + } + + // Mid-workflow encounter (rare) — park on a bookmark so a matching message can resume. + var stream = Stream.Get(context); + var subject = Subject.Get(context); + context.CreateBookmark(new NatsTriggerStimulus(stream, subject), OnResumeAsync, includeActivityInstanceId: false); + } + + private async ValueTask OnResumeAsync(ActivityExecutionContext context) + { + EmitOutputs(context); + await context.CompleteActivityAsync(); + } + + private void EmitOutputs(ActivityExecutionContext context) + { + var body = context.GetWorkflowInput(BodyInputKey) ?? string.Empty; + var receivedSubject = context.GetWorkflowInput(SubjectInputKey) ?? string.Empty; + var headers = context.GetWorkflowInput>(HeadersInputKey) ?? new Dictionary(); + + context.Set(Result, body); + context.Set(ReceivedSubject, receivedSubject); + context.Set(Headers, headers); + } +} diff --git a/src/Elsa.CustomServer/Elsa.CustomServer.csproj b/src/Elsa.CustomServer/Elsa.CustomServer.csproj index 38e3b37..7a5db53 100644 --- a/src/Elsa.CustomServer/Elsa.CustomServer.csproj +++ b/src/Elsa.CustomServer/Elsa.CustomServer.csproj @@ -20,6 +20,9 @@ + + + diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs b/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs new file mode 100644 index 0000000..9506415 --- /dev/null +++ b/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs @@ -0,0 +1,200 @@ +using System.Collections.Concurrent; +using System.Text; +using System.Text.Json; +using Elsa.CustomServer.Activities; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Filters; +using Elsa.Workflows.Runtime.Messages; +using Elsa.Workflows.Runtime.Requests; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using NATS.Client.Core; +using NATS.Client.JetStream; +using NATS.Client.JetStream.Models; + +namespace Elsa.CustomServer.Nats; + +// Consumes NATS JetStream on behalf of NatsTrigger-typed workflow triggers. +// On startup (and again whenever the trigger index changes) it groups the stored +// triggers by (stream, subject), spins up one durable pull consumer per unique +// pair, and dispatches a workflow per received message. Uses a stable, hashed +// DurableName that includes the stream + subject so multiple Elsa instances form +// a load-balanced consumer group for each subject. +public sealed class NatsTriggerHostedService( + INatsConnection nats, + IServiceProvider services, + ILogger log) + : BackgroundService +{ + private readonly ConcurrentDictionary _subscriptions = new(); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + try + { + await RefreshSubscriptionsAsync(stoppingToken); + + // Idle until cancellation. Refreshes happen out-of-band via NatsTriggerIndexChangedHandler. + await Task.Delay(Timeout.Infinite, stoppingToken); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + } + } + + public async Task RefreshSubscriptionsAsync(CancellationToken ct) + { + using var scope = services.CreateScope(); + var triggerStore = scope.ServiceProvider.GetRequiredService(); + var stored = await triggerStore.FindManyAsync(new TriggerFilter { Name = NatsTrigger.StimulusName }, ct); + + var desired = new HashSet(); + foreach (var t in stored) + { + var stim = TryDeserializeStimulus(t.Payload); + if (stim is null) + continue; + desired.Add(stim); + } + + // Stop consumers for triggers that are no longer indexed. + foreach (var key in _subscriptions.Keys.ToList()) + { + if (desired.Contains(key)) continue; + if (_subscriptions.TryRemove(key, out var cts)) + { + log.LogInformation("Stopping NATS consumer for stream={Stream} subject={Subject}", key.Stream, key.Subject); + cts.Cancel(); + cts.Dispose(); + } + } + + // Start consumers for newly indexed triggers. + foreach (var key in desired) + { + if (_subscriptions.ContainsKey(key)) continue; + var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + if (_subscriptions.TryAdd(key, cts)) + { + log.LogInformation("Starting NATS consumer for stream={Stream} subject={Subject}", key.Stream, key.Subject); + _ = Task.Run(() => ConsumeLoopAsync(key.Stream, key.Subject, cts.Token), CancellationToken.None); + } + } + } + + private static NatsTriggerStimulus? TryDeserializeStimulus(object? payload) + { + if (payload is null) return null; + if (payload is NatsTriggerStimulus s) return s; + + // Persistence round-trips the payload through JSON, so it typically comes back as JsonElement. + try + { + var json = payload as string ?? JsonSerializer.Serialize(payload); + return JsonSerializer.Deserialize(json, new JsonSerializerOptions + { + PropertyNameCaseInsensitive = true, + }); + } + catch + { + return null; + } + } + + private async Task ConsumeLoopAsync(string stream, string subject, CancellationToken ct) + { + // Deterministic consumer name so every Elsa instance joins the same group. + // Sanitize because NATS consumer names disallow '.', '*', '>', whitespace. + var consumerName = $"elsa-{Sanitize(stream)}-{Sanitize(subject)}"; + + var js = new NatsJSContext((NatsConnection)nats); + + // Reconnect-forever outer loop — if the consumer creation or consume stream + // faults, back off and try again. NatsConnection handles TCP-level reconnects + // itself; this catches higher-level errors (server restart, permissions). + while (!ct.IsCancellationRequested) + { + try + { + var consumer = await js.CreateOrUpdateConsumerAsync(stream, new ConsumerConfig + { + DurableName = consumerName, + AckPolicy = ConsumerConfigAckPolicy.Explicit, + DeliverPolicy = ConsumerConfigDeliverPolicy.New, // only applied on first-time creation + FilterSubject = subject, + MaxAckPending = 256, + AckWait = TimeSpan.FromSeconds(30), + MaxDeliver = 5, + }, ct); + + await foreach (var msg in consumer.ConsumeAsync(cancellationToken: ct)) + { + try + { + await DispatchAsync(stream, subject, msg, ct); + await msg.AckAsync(cancellationToken: ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + log.LogError(ex, "NATS dispatch failed for stream={Stream} subject={Subject} seq={Seq}; NAKing", stream, msg.Subject, msg.Metadata?.Sequence.Stream); + try + { + await msg.NakAsync(delay: TimeSpan.FromSeconds(5), cancellationToken: ct); + } + catch (Exception nakEx) + { + log.LogWarning(nakEx, "NAK failed for stream={Stream} subject={Subject}", stream, msg.Subject); + } + } + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + log.LogError(ex, "NATS consume loop crashed for stream={Stream} subject={Subject}; retrying in 5s", stream, subject); + try { await Task.Delay(TimeSpan.FromSeconds(5), ct); } + catch (OperationCanceledException) { return; } + } + } + } + + private async Task DispatchAsync(string stream, string subject, NatsJSMsg msg, CancellationToken ct) + { + using var scope = services.CreateScope(); + var dispatcher = scope.ServiceProvider.GetRequiredService(); + + var body = msg.Data is { Length: > 0 } bytes ? Encoding.UTF8.GetString(bytes) : string.Empty; + var headers = new Dictionary(); + if (msg.Headers is { } h) + { + foreach (var key in h.Keys) + headers[key] = h[key].ToString(); + } + + var request = new DispatchTriggerWorkflowsRequest(NatsTrigger.StimulusName, new NatsTriggerStimulus(stream, subject)) + { + Input = new Dictionary + { + [NatsTrigger.BodyInputKey] = body, + [NatsTrigger.SubjectInputKey] = msg.Subject, + [NatsTrigger.HeadersInputKey] = headers, + }, + }; + + await dispatcher.DispatchAsync(request, options: null!, cancellationToken: ct); + log.LogInformation("NATS → dispatched trigger stream={Stream} subject={Subject} seq={Seq}", + stream, msg.Subject, msg.Metadata?.Sequence.Stream); + } + + private static string Sanitize(string value) => + new string(value.Select(c => char.IsLetterOrDigit(c) ? c : '_').ToArray()); +} diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerIndexChangedHandler.cs b/src/Elsa.CustomServer/Nats/NatsTriggerIndexChangedHandler.cs new file mode 100644 index 0000000..85ce703 --- /dev/null +++ b/src/Elsa.CustomServer/Nats/NatsTriggerIndexChangedHandler.cs @@ -0,0 +1,13 @@ +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Runtime.Notifications; + +namespace Elsa.CustomServer.Nats; + +// Refreshes the NATS subscriber whenever workflow triggers are re-indexed +// (e.g. after publishing a workflow that adds or removes a NatsTrigger). +public sealed class NatsTriggerIndexChangedHandler(NatsTriggerHostedService subscriber) + : INotificationHandler +{ + public Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken) => + subscriber.RefreshSubscriptionsAsync(cancellationToken); +} diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs b/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs new file mode 100644 index 0000000..eb91e9f --- /dev/null +++ b/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs @@ -0,0 +1,6 @@ +namespace Elsa.CustomServer.Nats; + +// Stimulus used both as the indexed trigger payload and as the bookmark payload +// when a NATS message dispatches a workflow. Structural equality (record) is what +// lets the runtime match published triggers against incoming messages. +public sealed record NatsTriggerStimulus(string Stream, string Subject); diff --git a/src/Elsa.CustomServer/Program.cs b/src/Elsa.CustomServer/Program.cs index 887b125..47589bd 100644 --- a/src/Elsa.CustomServer/Program.cs +++ b/src/Elsa.CustomServer/Program.cs @@ -1,10 +1,14 @@ using System.Text.Encodings.Web; +using Elsa.CustomServer.Nats; using Elsa.EntityFrameworkCore.Extensions; using Elsa.EntityFrameworkCore.Modules.Management; using Elsa.EntityFrameworkCore.Modules.Runtime; using Elsa.Extensions; +using Elsa.Mediator.Extensions; using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.HttpOverrides; +using NATS.Client.Core; +using NATS.Client.Hosting; var builder = WebApplication.CreateBuilder(args); builder.WebHost.UseStaticWebAssets(); @@ -38,6 +42,34 @@ builder.Services.AddElsa(elsa => .UseLiquid(liquid => liquid.FluidOptions = options => options.Encoder = HtmlEncoder.Default); }); +// NATS: single shared NatsConnection wired from Nats:Url / Nats:User / Nats:Password +// (with env-var fallback NATS_URL / NATS_USER / NATS_PASSWORD). The AddNats hosting +// extension registers INatsConnection as a singleton; wss://, tls://, nats:// are +// all handled transparently based on the URL scheme. +var natsSection = configuration.GetSection("Nats"); +var natsUrl = natsSection["Url"] ?? Environment.GetEnvironmentVariable("NATS_URL"); +var natsUser = natsSection["User"] ?? Environment.GetEnvironmentVariable("NATS_USER"); +var natsPassword = natsSection["Password"] ?? Environment.GetEnvironmentVariable("NATS_PASSWORD"); + +if (!string.IsNullOrWhiteSpace(natsUrl)) +{ + builder.Services.AddNats(configureOpts: opts => opts with + { + Url = natsUrl, + Name = "elsa-workflows", + AuthOpts = new NatsAuthOpts + { + Username = string.IsNullOrEmpty(natsUser) ? null : natsUser, + Password = string.IsNullOrEmpty(natsPassword) ? null : natsPassword, + }, + }); + + // Hosted subscriber that fans workflow-trigger dispatches out from stored NatsTrigger definitions. + builder.Services.AddSingleton(); + builder.Services.AddHostedService(sp => sp.GetRequiredService()); + builder.Services.AddNotificationHandler(); +} + builder.Services.AddCors(cors => cors.AddDefaultPolicy(p => p .AllowAnyHeader() .AllowAnyMethod() diff --git a/src/Elsa.CustomServer/appsettings.json b/src/Elsa.CustomServer/appsettings.json index 8ef4081..eee83f7 100644 --- a/src/Elsa.CustomServer/appsettings.json +++ b/src/Elsa.CustomServer/appsettings.json @@ -16,6 +16,11 @@ "Http": { "BasePath": "/workflows" }, + "Nats": { + "Url": "", + "User": "", + "Password": "" + }, "Identity": { "Tokens": { "SigningKey": "CHANGE_ME_VIA_ENV_Identity__Tokens__SigningKey",