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