From b2c281193317fb6bc9c0891328847610eb9e38d3 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Wed, 2 Sep 2026 09:57:04 +0000 Subject: [PATCH] Move NATS connection details from env vars onto the activity nodes Url/User/Password are now inputs on NatsPublish and NatsTrigger so authors configure them in the workflow designer. The stored trigger stimulus carries the connection info too, so messages from server A only fire workflows designed to listen on server A. Dropped the AddNats singleton wiring and the Nats section from appsettings. A NatsConnectionCache singleton hands out one NatsConnection per unique (url, user, password) so repeat publishes and each hosted-service consumer loop reuse the same TCP session. --- CLAUDE.md | 49 +++++++++-------- .../Activities/NatsPublish.cs | 19 ++++++- .../Activities/NatsTrigger.cs | 28 +++++++--- .../Elsa.CustomServer.csproj | 4 +- .../Nats/NatsConnectionCache.cs | 46 ++++++++++++++++ .../Nats/NatsTriggerHostedService.cs | 53 +++++++++---------- .../Nats/NatsTriggerStimulus.cs | 15 +++++- src/Elsa.CustomServer/Program.cs | 38 ++++--------- src/Elsa.CustomServer/appsettings.json | 5 -- 9 files changed, 161 insertions(+), 96 deletions(-) create mode 100644 src/Elsa.CustomServer/Nats/NatsConnectionCache.cs diff --git a/CLAUDE.md b/CLAUDE.md index 2b0ce4f..52de845 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -171,23 +171,32 @@ to its DSN. ## NATS activities -Three files under `src/Elsa.CustomServer/`: +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. -- `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`. +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 + `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 + designed to listen on server A. +- `Nats/NatsConnectionCache.cs` — singleton `(url, user, password) → NatsConnection` cache. - `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). + durable pull consumer per unique stimulus, 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 @@ -195,14 +204,12 @@ 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. +**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. -Packages: `NATS.Net` + `NATS.Client.Hosting` (v2 client). The v1 `NATS.Client` -package does not support `wss://` from .NET — do not use it. +Package: `NATS.Net` (v2 client). The v1 `NATS.Client` package does not support +`wss://` from .NET — do not use it. ## Session-specific reminders diff --git a/src/Elsa.CustomServer/Activities/NatsPublish.cs b/src/Elsa.CustomServer/Activities/NatsPublish.cs index 509af1c..5233c1f 100644 --- a/src/Elsa.CustomServer/Activities/NatsPublish.cs +++ b/src/Elsa.CustomServer/Activities/NatsPublish.cs @@ -1,4 +1,5 @@ using System.Text; +using Elsa.CustomServer.Nats; using Elsa.Extensions; using Elsa.Workflows; using Elsa.Workflows.Attributes; @@ -12,6 +13,15 @@ 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(Description = "Subject to publish to (must be bound to a JetStream stream).")] public Input Subject { get; set; } = default!; @@ -32,6 +42,10 @@ 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 subject = context.Get(Subject); if (string.IsNullOrWhiteSpace(subject)) throw new InvalidOperationException("NatsPublish: Subject is required."); @@ -40,8 +54,9 @@ public class NatsPublish : CodeActivity var messageId = context.Get(MessageId); var expectedStream = context.Get(ExpectedStream); - var nats = context.GetRequiredService(); - var js = new NatsJSContext((NatsConnection)nats); + var pool = context.GetRequiredService(); + var nats = pool.Get(url, context.Get(User), context.Get(Password)); + var js = new NatsJSContext(nats); NatsHeaders? headers = null; if (!string.IsNullOrEmpty(messageId)) diff --git a/src/Elsa.CustomServer/Activities/NatsTrigger.cs b/src/Elsa.CustomServer/Activities/NatsTrigger.cs index 34f3b5e..0c3d8ab 100644 --- a/src/Elsa.CustomServer/Activities/NatsTrigger.cs +++ b/src/Elsa.CustomServer/Activities/NatsTrigger.cs @@ -1,4 +1,5 @@ using Elsa.CustomServer.Nats; +using Elsa.Expressions.Models; using Elsa.Extensions; using Elsa.Workflows; using Elsa.Workflows.Attributes; @@ -18,10 +19,19 @@ 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(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.")] + [Input(Description = "Subject to subscribe to. Wildcards 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).")] @@ -33,9 +43,7 @@ public class NatsTrigger : Trigger 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)]; + return [BuildStimulus(context.ExpressionExecutionContext)]; } protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) @@ -48,11 +56,17 @@ public class NatsTrigger : Trigger } // 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); + context.CreateBookmark(BuildStimulus(context.ExpressionExecutionContext), OnResumeAsync, includeActivityInstanceId: false); } + private NatsTriggerStimulus BuildStimulus(ExpressionExecutionContext context) => + new( + Url.Get(context) ?? string.Empty, + User.Get(context) ?? string.Empty, + Password.Get(context) ?? string.Empty, + Stream.Get(context) ?? string.Empty, + Subject.Get(context) ?? string.Empty); + private async ValueTask OnResumeAsync(ActivityExecutionContext context) { EmitOutputs(context); diff --git a/src/Elsa.CustomServer/Elsa.CustomServer.csproj b/src/Elsa.CustomServer/Elsa.CustomServer.csproj index 7a5db53..5732cc5 100644 --- a/src/Elsa.CustomServer/Elsa.CustomServer.csproj +++ b/src/Elsa.CustomServer/Elsa.CustomServer.csproj @@ -20,9 +20,9 @@ - + - diff --git a/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs b/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs new file mode 100644 index 0000000..845f7e4 --- /dev/null +++ b/src/Elsa.CustomServer/Nats/NatsConnectionCache.cs @@ -0,0 +1,46 @@ +using System.Collections.Concurrent; +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. +public sealed class NatsConnectionCache : IAsyncDisposable +{ + private readonly ConcurrentDictionary> _connections = new(); + + public NatsConnection Get(string url, string? user, string? password) + { + if (string.IsNullOrWhiteSpace(url)) + throw new ArgumentException("NATS Url is required.", nameof(url)); + + var key = $"{url}|{user}|{password}"; + var lazy = _connections.GetOrAdd(key, _ => new Lazy(() => + { + var opts = NatsOpts.Default with + { + Url = url, + Name = "elsa-workflows", + AuthOpts = new NatsAuthOpts + { + Username = string.IsNullOrEmpty(user) ? null : user, + Password = string.IsNullOrEmpty(password) ? null : password, + }, + }; + return new NatsConnection(opts); + })); + + return lazy.Value; + } + + public async ValueTask DisposeAsync() + { + foreach (var entry in _connections.Values) + { + if (!entry.IsValueCreated) continue; + try { await entry.Value.DisposeAsync(); } catch { /* best-effort */ } + } + _connections.Clear(); + } +} diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs b/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs index 9506415..dd66a49 100644 --- a/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs +++ b/src/Elsa.CustomServer/Nats/NatsTriggerHostedService.cs @@ -4,25 +4,22 @@ 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. +// 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 +// DurableName so multiple Elsa instances form a load-balanced group. public sealed class NatsTriggerHostedService( - INatsConnection nats, + NatsConnectionCache connectionCache, IServiceProvider services, ILogger log) : BackgroundService @@ -53,7 +50,7 @@ public sealed class NatsTriggerHostedService( foreach (var t in stored) { var stim = TryDeserializeStimulus(t.Payload); - if (stim is null) + if (stim is null || string.IsNullOrWhiteSpace(stim.Url) || string.IsNullOrWhiteSpace(stim.Stream) || string.IsNullOrWhiteSpace(stim.Subject)) continue; desired.Add(stim); } @@ -64,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 stream={Stream} subject={Subject}", key.Stream, key.Subject); + log.LogInformation("Stopping NATS consumer for url={Url} stream={Stream} subject={Subject}", key.Url, key.Stream, key.Subject); cts.Cancel(); cts.Dispose(); } @@ -77,8 +74,8 @@ public sealed class NatsTriggerHostedService( 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); + log.LogInformation("Starting NATS consumer for url={Url} stream={Stream} subject={Subject}", key.Url, key.Stream, key.Subject); + _ = Task.Run(() => ConsumeLoopAsync(key, cts.Token), CancellationToken.None); } } } @@ -103,27 +100,27 @@ public sealed class NatsTriggerHostedService( } } - private async Task ConsumeLoopAsync(string stream, string subject, CancellationToken ct) + private async Task ConsumeLoopAsync(NatsTriggerStimulus stim, 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 consumerName = $"elsa-{Sanitize(stim.Stream)}-{Sanitize(stim.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). + // Reconnect-forever outer loop — catches higher-level errors (server restart, permissions). + // NatsConnection handles TCP-level reconnects itself. while (!ct.IsCancellationRequested) { try { - var consumer = await js.CreateOrUpdateConsumerAsync(stream, new ConsumerConfig + var nats = connectionCache.Get(stim.Url, stim.User, stim.Password); + var js = new NatsJSContext(nats); + + var consumer = await js.CreateOrUpdateConsumerAsync(stim.Stream, new ConsumerConfig { DurableName = consumerName, AckPolicy = ConsumerConfigAckPolicy.Explicit, DeliverPolicy = ConsumerConfigDeliverPolicy.New, // only applied on first-time creation - FilterSubject = subject, + FilterSubject = stim.Subject, MaxAckPending = 256, AckWait = TimeSpan.FromSeconds(30), MaxDeliver = 5, @@ -133,7 +130,7 @@ public sealed class NatsTriggerHostedService( { try { - await DispatchAsync(stream, subject, msg, ct); + await DispatchAsync(stim, msg, ct); await msg.AckAsync(cancellationToken: ct); } catch (OperationCanceledException) when (ct.IsCancellationRequested) @@ -142,14 +139,14 @@ public sealed class NatsTriggerHostedService( } catch (Exception ex) { - log.LogError(ex, "NATS dispatch failed for stream={Stream} subject={Subject} seq={Seq}; NAKing", stream, msg.Subject, msg.Metadata?.Sequence.Stream); + log.LogError(ex, "NATS dispatch failed for stream={Stream} subject={Subject} seq={Seq}; NAKing", stim.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); + log.LogWarning(nakEx, "NAK failed for stream={Stream} subject={Subject}", stim.Stream, msg.Subject); } } } @@ -160,14 +157,14 @@ public sealed class NatsTriggerHostedService( } catch (Exception ex) { - log.LogError(ex, "NATS consume loop crashed for stream={Stream} subject={Subject}; retrying in 5s", stream, subject); + log.LogError(ex, "NATS consume loop crashed for url={Url} stream={Stream} subject={Subject}; retrying in 5s", stim.Url, stim.Stream, stim.Subject); try { await Task.Delay(TimeSpan.FromSeconds(5), ct); } catch (OperationCanceledException) { return; } } } } - private async Task DispatchAsync(string stream, string subject, NatsJSMsg msg, CancellationToken ct) + private async Task DispatchAsync(NatsTriggerStimulus stim, NatsJSMsg msg, CancellationToken ct) { using var scope = services.CreateScope(); var dispatcher = scope.ServiceProvider.GetRequiredService(); @@ -180,7 +177,7 @@ public sealed class NatsTriggerHostedService( headers[key] = h[key].ToString(); } - var request = new DispatchTriggerWorkflowsRequest(NatsTrigger.StimulusName, new NatsTriggerStimulus(stream, subject)) + var request = new DispatchTriggerWorkflowsRequest(NatsTrigger.StimulusName, stim) { Input = new Dictionary { @@ -192,7 +189,7 @@ public sealed class NatsTriggerHostedService( 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); + stim.Stream, msg.Subject, msg.Metadata?.Sequence.Stream); } private static string Sanitize(string value) => diff --git a/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs b/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs index eb91e9f..76b387e 100644 --- a/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs +++ b/src/Elsa.CustomServer/Nats/NatsTriggerStimulus.cs @@ -2,5 +2,16 @@ 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); +// 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 +// 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. +public sealed record NatsTriggerStimulus( + string Url, + string User, + string Password, + string Stream, + string Subject); diff --git a/src/Elsa.CustomServer/Program.cs b/src/Elsa.CustomServer/Program.cs index 47589bd..efaf716 100644 --- a/src/Elsa.CustomServer/Program.cs +++ b/src/Elsa.CustomServer/Program.cs @@ -7,8 +7,6 @@ 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(); @@ -42,33 +40,15 @@ 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(); -} +// NATS: connection details (URL/user/password) live on the NatsPublish / NatsTrigger +// activity nodes themselves — configured in the workflow designer. A single pool +// caches one NatsConnection per (url, user, password) so repeat invocations reuse +// the same connection. The hosted service reads stored NatsTrigger stimuli and +// spins up one durable pull consumer per unique connection+subject. +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddHostedService(sp => sp.GetRequiredService()); +builder.Services.AddNotificationHandler(); builder.Services.AddCors(cors => cors.AddDefaultPolicy(p => p .AllowAnyHeader() diff --git a/src/Elsa.CustomServer/appsettings.json b/src/Elsa.CustomServer/appsettings.json index eee83f7..8ef4081 100644 --- a/src/Elsa.CustomServer/appsettings.json +++ b/src/Elsa.CustomServer/appsettings.json @@ -16,11 +16,6 @@ "Http": { "BasePath": "/workflows" }, - "Nats": { - "Url": "", - "User": "", - "Password": "" - }, "Identity": { "Tokens": { "SigningKey": "CHANGE_ME_VIA_ENV_Identity__Tokens__SigningKey",