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<string>)
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

@@ -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<string>`. 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-<sanitizedStream>-<sanitizedSubject>` 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

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<string> Subject { get; set; } = default!;
+
+    [Input(Description = "Message body (UTF-8 string).")]
+    public Input<string> 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<string?> 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<string?> ExpectedStream { get; set; } = new(default(string?));
+
+    [Output(Description = "Stream sequence number the server assigned to the accepted message.")]
+    public Output<long> Sequence { get; set; } = default!;
+
+    [Output(Description = "True when JetStream deduplicated (Nats-Msg-Id already seen within the duplicate window).")]
+    public Output<bool> 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<INatsConnection>();
+        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);
+    }
+}

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<string>
+{
+    // 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<string> Stream { get; set; } = default!;
+
+    [Input(Description = "Subject to subscribe to. Wildcards are allowed: 'orders.*' matches one token, 'orders.>' matches one or more.")]
+    public Input<string> Subject { get; set; } = default!;
+
+    [Output(Description = "The concrete subject the message was published to (useful with wildcards).")]
+    public Output<string> ReceivedSubject { get; set; } = default!;
+
+    [Output(Description = "Message headers as a string dictionary.")]
+    public Output<IDictionary<string, string>> Headers { get; set; } = default!;
+
+    protected override IEnumerable<object> 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<string>(BodyInputKey) ?? string.Empty;
+        var receivedSubject = context.GetWorkflowInput<string>(SubjectInputKey) ?? string.Empty;
+        var headers = context.GetWorkflowInput<IDictionary<string, string>>(HeadersInputKey) ?? new Dictionary<string, string>();
+
+        context.Set(Result, body);
+        context.Set(ReceivedSubject, receivedSubject);
+        context.Set(Headers, headers);
+    }
+}

src/Elsa.CustomServer/Elsa.CustomServer.csproj

@@ -20,6 +20,9 @@
     <PackageReference Include="Elsa.CSharp" Version="3.5.3" />
     <PackageReference Include="Elsa.JavaScript" Version="3.5.3" />
     <PackageReference Include="Elsa.Liquid" Version="3.5.3" />
+    <!-- NATS.Net v2 client + hosting extensions for JetStream publish/subscribe. -->
+    <PackageReference Include="NATS.Net" Version="2.5.7" />
+    <PackageReference Include="NATS.Client.Hosting" Version="2.5.7" />
     <!-- Hosts the Blazor WebAssembly Studio in the same process. -->
     <PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.13" />
   </ItemGroup>

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<NatsTriggerHostedService> log)
+    : BackgroundService
+{
+    private readonly ConcurrentDictionary<NatsTriggerStimulus, CancellationTokenSource> _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<ITriggerStore>();
+        var stored = await triggerStore.FindManyAsync(new TriggerFilter { Name = NatsTrigger.StimulusName }, ct);
+
+        var desired = new HashSet<NatsTriggerStimulus>();
+        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<NatsTriggerStimulus>(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<byte[]>(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<byte[]> msg, CancellationToken ct)
+    {
+        using var scope = services.CreateScope();
+        var dispatcher = scope.ServiceProvider.GetRequiredService<IWorkflowDispatcher>();
+
+        var body = msg.Data is { Length: > 0 } bytes ? Encoding.UTF8.GetString(bytes) : string.Empty;
+        var headers = new Dictionary<string, string>();
+        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<string, object>
+            {
+                [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());
+}

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<WorkflowTriggersIndexed>
+{
+    public Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken) =>
+        subscriber.RefreshSubscriptionsAsync(cancellationToken);
+}

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);

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<NatsTriggerHostedService>();
+    builder.Services.AddHostedService(sp => sp.GetRequiredService<NatsTriggerHostedService>());
+    builder.Services.AddNotificationHandler<NatsTriggerIndexChangedHandler, Elsa.Workflows.Runtime.Notifications.WorkflowTriggersIndexed>();
+}
+
 builder.Services.AddCors(cors => cors.AddDefaultPolicy(p => p
     .AllowAnyHeader()
     .AllowAnyMethod()

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",