From 2378a5ff78a91c128accd0e9325a338cbaa2cf65 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Wed, 2 Sep 2026 14:56:59 +0000 Subject: [PATCH] Collapse NATS activity credentials into a single Connection string Replaces separate Url/User/Password inputs on NatsPublish and NatsTrigger with one Connection input in standard NATS URL form (e.g. nats://user:pass@host:4222, wss://user:pass@host/). The activity's input description explains the format so it renders as help text in the Studio property panel. NatsConnectionCache now parses the connection string, keeps its cache keyed on the parsed (url, user, password) tuple, and exposes ParseConnectionString for reuse. NatsTriggerStimulus collapses to (Connection, Stream, Subject); existing NATS workflows must be republished to re-index with the new stimulus shape. --- CLAUDE.md | 49 +++++++++++------- .../Activities/NatsPublish.cs | 26 +++++----- .../Activities/NatsTrigger.cs | 22 ++++---- .../Nats/NatsConnectionCache.cs | 51 ++++++++++++++++--- .../Nats/NatsTriggerHostedService.cs | 16 +++--- .../Nats/NatsTriggerStimulus.cs | 11 ++-- 6 files changed, 114 insertions(+), 61 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 52de845..67b7c6d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -171,25 +171,33 @@ to its DSN. ## NATS activities -Connection details (URL, user, password) live on the activity nodes themselves -— configured per-node in the workflow designer, not in appsettings or env vars. -Every stored trigger carries its own connection info; a `NatsConnectionCache` -singleton hands out one `NatsConnection` per unique `(url, user, password)` so -repeat invocations reuse the same TCP/TLS session. +Connection details live on the activity nodes themselves — configured per-node +in the workflow designer as a **single connection string** in the standard NATS +URL form `://[user:password@]host[:port]` (schemes: `nats`, `tls`, +`ws`, `wss`). Every stored trigger carries its own connection string; a +`NatsConnectionCache` singleton parses the string into `(cleanUrl, user, +password)` and hands out one `NatsConnection` per unique tuple so repeat +invocations reuse the same TCP/TLS session. + +Passwords with special characters (`@ : / #`) must be URL-encoded (`%40 %3A +%2F %23`). The activity's input `Description` shows this in the Studio +property panel. Files under `src/Elsa.CustomServer/`: -- `Activities/NatsPublish.cs` — publisher `CodeActivity`. Inputs: `Url`, - `User`, `Password`, `Subject`, `Body`, optional `MessageId` (sets the - `Nats-Msg-Id` header for JetStream server-side dedup), optional +- `Activities/NatsPublish.cs` — publisher `CodeActivity`. Inputs: `Connection` + (single connection string), `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: `Url`, - `User`, `Password`, `Stream`, `Subject` (wildcards allowed). Outputs: - `Result` = message body (UTF-8), `ReceivedSubject`, `Headers`. Its trigger - name / stimulus name is `NATS.NatsTrigger`; the stimulus record includes the - connection info so a message from server A only starts the workflows +- `Activities/NatsTrigger.cs` — trigger `Trigger`. Inputs: + `Connection`, `Stream`, `Subject` (wildcards allowed). Outputs: `Result` = + message body (UTF-8), `ReceivedSubject`, `Headers`. Its trigger name / + stimulus name is `NATS.NatsTrigger`; the stimulus record includes the + connection string so a message from server A only starts the workflows designed to listen on server A. -- `Nats/NatsConnectionCache.cs` — singleton `(url, user, password) → NatsConnection` cache. +- `Nats/NatsConnectionCache.cs` — singleton connection-string parser + + `(cleanUrl, user, password) → NatsConnection` cache. Exposes static + `ParseConnectionString(...)` for anyone else who needs the same split. - `Nats/NatsTriggerHostedService.cs` — one `BackgroundService` that scans `ITriggerStore` for stored triggers named `NATS.NatsTrigger`, spins up one durable pull consumer per unique stimulus, and dispatches workflows via @@ -204,9 +212,16 @@ 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. -**Security note:** trigger stimuli — including the NATS password — are -persisted in the trigger store DB. Don't log the stimulus record verbatim, and -treat the trigger store with the same care as any secret storage. +**Security note:** trigger stimuli — including the NATS connection string, +which typically embeds the password — are persisted in the trigger store DB. +Don't log the stimulus record verbatim, and treat the trigger store with the +same care as any secret storage. + +**Migration note:** activities were changed from three fields +(`Url / User / Password`) to a single `Connection` input. Any workflow authored +against the old shape needs to be reopened in Studio and its NATS +publish/trigger nodes re-filled — republishing then re-indexes the trigger +store with the new stimulus shape. Package: `NATS.Net` (v2 client). The v1 `NATS.Client` package does not support `wss://` from .NET — do not use it. diff --git a/src/Elsa.CustomServer/Activities/NatsPublish.cs b/src/Elsa.CustomServer/Activities/NatsPublish.cs index 5233c1f..9cd93cd 100644 --- a/src/Elsa.CustomServer/Activities/NatsPublish.cs +++ b/src/Elsa.CustomServer/Activities/NatsPublish.cs @@ -13,14 +13,16 @@ 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 = "NATS server URL (nats://, tls://, ws://, wss://).", Category = "Connection")] - public Input Url { get; set; } = default!; - - [Input(Description = "Username (optional).", Category = "Connection")] - public Input User { get; set; } = new(default(string?)); - - [Input(Description = "Password (optional).", Category = "Connection")] - public Input Password { get; set; } = new(default(string?)); + [Input( + DisplayName = "Connection string", + Description = + "NATS connection string. Format: ://[user:password@]host[:port]. " + + "Schemes: nats, tls, ws, wss. " + + "Examples: nats://demo.nats.io:4222 | nats://alice:s3cret@nats.example.com:4222 | " + + "wss://alice:s3cret@nats.example.com/. " + + "Special characters in the password must be URL-encoded (@ = %40, : = %3A, / = %2F, # = %23).", + Category = "Connection")] + public Input Connection { get; set; } = default!; [Input(Description = "Subject to publish to (must be bound to a JetStream stream).")] public Input Subject { get; set; } = default!; @@ -42,9 +44,9 @@ public class NatsPublish : CodeActivity protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) { - var url = context.Get(Url); - if (string.IsNullOrWhiteSpace(url)) - throw new InvalidOperationException("NatsPublish: Url is required."); + var connection = context.Get(Connection); + if (string.IsNullOrWhiteSpace(connection)) + throw new InvalidOperationException("NatsPublish: Connection string is required."); var subject = context.Get(Subject); if (string.IsNullOrWhiteSpace(subject)) @@ -55,7 +57,7 @@ public class NatsPublish : CodeActivity var expectedStream = context.Get(ExpectedStream); var pool = context.GetRequiredService(); - var nats = pool.Get(url, context.Get(User), context.Get(Password)); + var nats = pool.Get(connection); var js = new NatsJSContext(nats); NatsHeaders? headers = null; diff --git a/src/Elsa.CustomServer/Activities/NatsTrigger.cs b/src/Elsa.CustomServer/Activities/NatsTrigger.cs index 0c3d8ab..c080002 100644 --- a/src/Elsa.CustomServer/Activities/NatsTrigger.cs +++ b/src/Elsa.CustomServer/Activities/NatsTrigger.cs @@ -19,14 +19,16 @@ public class NatsTrigger : Trigger public const string SubjectInputKey = "NatsSubject"; public const string HeadersInputKey = "NatsHeaders"; - [Input(Description = "NATS server URL (nats://, tls://, ws://, wss://).", Category = "Connection")] - public Input Url { get; set; } = default!; - - [Input(Description = "Username (optional).", Category = "Connection")] - public Input User { get; set; } = new(default(string?)); - - [Input(Description = "Password (optional).", Category = "Connection")] - public Input Password { get; set; } = new(default(string?)); + [Input( + DisplayName = "Connection string", + Description = + "NATS connection string. Format: ://[user:password@]host[:port]. " + + "Schemes: nats, tls, ws, wss. " + + "Examples: nats://demo.nats.io:4222 | nats://alice:s3cret@nats.example.com:4222 | " + + "wss://alice:s3cret@nats.example.com/. " + + "Special characters in the password must be URL-encoded (@ = %40, : = %3A, / = %2F, # = %23).", + Category = "Connection")] + public Input Connection { get; set; } = default!; [Input(Description = "JetStream name that owns the subject.")] public Input Stream { get; set; } = default!; @@ -61,9 +63,7 @@ public class NatsTrigger : Trigger private NatsTriggerStimulus BuildStimulus(ExpressionExecutionContext context) => new( - Url.Get(context) ?? string.Empty, - User.Get(context) ?? string.Empty, - Password.Get(context) ?? string.Empty, + Connection.Get(context) ?? string.Empty, Stream.Get(context) ?? string.Empty, Subject.Get(context) ?? string.Empty); diff --git a/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs b/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs index 845f7e4..e0fa239 100644 --- a/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs +++ b/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs @@ -3,17 +3,20 @@ using NATS.Client.Core; namespace Elsa.CustomServer.Nats; -// Lazily creates one NatsConnection per (url, user, password) tuple and reuses -// it for the process lifetime. Registered as a singleton so both the publish -// activity and the trigger subscriber share the same pool. +// Lazily creates one NatsConnection per parsed (url, user, password) tuple and +// reuses it for the process lifetime. Registered as a singleton so both the +// publish activity and the trigger subscriber share the same pool. public sealed class NatsConnectionCache : IAsyncDisposable { private readonly ConcurrentDictionary> _connections = new(); - public NatsConnection Get(string url, string? user, string? password) + // Accepts a single connection string of the form + // nats://user:pass@host:4222 or wss://user:pass@host/ (ws/tls also supported) + // Userinfo (`user:pass@`) is optional. Special characters in the password + // must be URL-encoded — `@` → `%40`, `:` → `%3A`, `/` → `%2F`, `#` → `%23`. + public NatsConnection Get(string connectionString) { - if (string.IsNullOrWhiteSpace(url)) - throw new ArgumentException("NATS Url is required.", nameof(url)); + var (url, user, password) = ParseConnectionString(connectionString); var key = $"{url}|{user}|{password}"; var lazy = _connections.GetOrAdd(key, _ => new Lazy(() => @@ -34,6 +37,42 @@ public sealed class NatsConnectionCache : IAsyncDisposable return lazy.Value; } + // Split a NATS connection string into (urlWithoutUserinfo, user, password). + // `Uri` decodes percent-escapes for us, so callers get the raw password. + public static (string Url, string? User, string? Password) ParseConnectionString(string connectionString) + { + if (string.IsNullOrWhiteSpace(connectionString)) + throw new ArgumentException("NATS connection string is required.", nameof(connectionString)); + + Uri uri; + try + { + uri = new Uri(connectionString, UriKind.Absolute); + } + catch (UriFormatException ex) + { + throw new ArgumentException( + $"NATS connection string is not a valid URL: '{connectionString}'. Expected e.g. nats://user:pass@host:4222", + nameof(connectionString), ex); + } + + string? user = null; + string? password = null; + if (!string.IsNullOrEmpty(uri.UserInfo)) + { + var parts = uri.UserInfo.Split(':', 2); + user = Uri.UnescapeDataString(parts[0]); + if (parts.Length == 2) + password = Uri.UnescapeDataString(parts[1]); + } + + // Rebuild the URL without userinfo so it can be handed to NatsOpts.Url cleanly. + var builder = new UriBuilder(uri) { UserName = string.Empty, Password = string.Empty }; + var cleanUrl = builder.Uri.ToString(); + + return (cleanUrl, user, password); + } + public async ValueTask DisposeAsync() { foreach (var entry in _connections.Values) diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs b/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs index ffb1951..9de9f9f 100644 --- a/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs +++ b/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs @@ -14,9 +14,9 @@ using NATS.Client.JetStream.Models; namespace Elsa.CustomServer.Nats; // Consumes NATS JetStream on behalf of NatsTrigger-typed workflow triggers. -// Each stored trigger carries its own (Url, User, Password, Stream, Subject) -// stimulus; the service spins up one durable pull consumer per unique stimulus -// and dispatches a workflow per received message. Consumers use a deterministic +// Each stored trigger carries its own (Connection, Stream, Subject) stimulus; +// the service spins up one durable pull consumer per unique stimulus and +// dispatches a workflow per received message. Consumers use a deterministic // DurableName so multiple Elsa instances form a load-balanced group. public sealed class NatsTriggerHostedService( NatsConnectionCache connectionCache, @@ -50,7 +50,7 @@ public sealed class NatsTriggerHostedService( foreach (var t in stored) { var stim = TryDeserializeStimulus(t.Payload); - if (stim is null || string.IsNullOrWhiteSpace(stim.Url) || string.IsNullOrWhiteSpace(stim.Stream) || string.IsNullOrWhiteSpace(stim.Subject)) + if (stim is null || string.IsNullOrWhiteSpace(stim.Connection) || string.IsNullOrWhiteSpace(stim.Stream) || string.IsNullOrWhiteSpace(stim.Subject)) continue; desired.Add(stim); } @@ -61,7 +61,7 @@ public sealed class NatsTriggerHostedService( if (desired.Contains(key)) continue; if (_subscriptions.TryRemove(key, out var cts)) { - log.LogInformation("Stopping NATS consumer for url={Url} stream={Stream} subject={Subject}", key.Url, key.Stream, key.Subject); + log.LogInformation("Stopping NATS consumer for stream={Stream} subject={Subject}", key.Stream, key.Subject); cts.Cancel(); cts.Dispose(); } @@ -74,7 +74,7 @@ public sealed class NatsTriggerHostedService( var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); if (_subscriptions.TryAdd(key, cts)) { - log.LogInformation("Starting NATS consumer for url={Url} stream={Stream} subject={Subject}", key.Url, key.Stream, key.Subject); + log.LogInformation("Starting NATS consumer for stream={Stream} subject={Subject}", key.Stream, key.Subject); _ = Task.Run(() => ConsumeLoopAsync(key, cts.Token), CancellationToken.None); } } @@ -112,7 +112,7 @@ public sealed class NatsTriggerHostedService( { try { - var nats = connectionCache.Get(stim.Url, stim.User, stim.Password); + var nats = connectionCache.Get(stim.Connection); var js = new NatsJSContext(nats); var consumer = await js.CreateOrUpdateConsumerAsync(stim.Stream, new ConsumerConfig @@ -162,7 +162,7 @@ public sealed class NatsTriggerHostedService( } catch (Exception ex) { - log.LogError(ex, "NATS consume loop crashed for url={Url} stream={Stream} subject={Subject}; retrying in 5s", stim.Url, stim.Stream, stim.Subject); + log.LogError(ex, "NATS consume loop crashed for stream={Stream} subject={Subject}; retrying in 5s", stim.Stream, stim.Subject); try { await Task.Delay(TimeSpan.FromSeconds(5), ct); } catch (OperationCanceledException) { return; } } diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs b/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs index 76b387e..ae6cb38 100644 --- a/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs +++ b/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs @@ -3,15 +3,12 @@ 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, so the -// connection details are part of the key: a message pulled from server A doesn't +// connection string is part of the key: a message pulled from server A doesn't // start a workflow that was designed to listen on server B. // -// Note: credentials are persisted in the trigger store as part of this payload, -// which is the trade-off for letting workflow authors configure NATS in the -// designer. Don't log this record verbatim. +// Note: the connection string may embed credentials and is persisted in the +// trigger store. Don't log this record verbatim. public sealed record NatsTriggerStimulus( - string Url, - string User, - string Password, + string Connection, string Stream, string Subject);