feat: add NATS / JetStream messaging gateway

[lillo42]

Summary

Adds a NATS transport for Brighter, covering both core NATS and JetStream.

What's included

• New Paramore.Brighter.MessagingGateway.NATS project:
  • Core NATS: NatsMessageProducer, NatsMessageConsumer, NatsSubscription, NatsPublication, NatsChannelFactory
  • JetStream: NatsStreamMessageProducer, NatsStreamMessageConsumer, NatsStreamSubscription, NatsStreamPublication
  • Producer/consumer factories and producer registry factory
  • Header mapping (NatsMsgExtensions, MessageHeaderExtensions, HeadersName) and NatsNameSanitizer
  • NatsMessageGatewayConfiguration
• Observability: MessagingSystem.NATS entries
• New Paramore.Brighter.NATS.Tests project (reactor + proactor gateway tests: send/receive, multiple messages, concurrent senders, failed-message redelivery, activity-context propagation)
• docker-compose-nats.yaml for local testing
• CI: nats-ci job running the NATS transport tests against a nats:latest service container

[claude]

Code Review — NATS / JetStream messaging gateway (PR #4249)

Thanks for this — it's a large, well-documented, and largely well-structured transport. The XML docs are excellent, the licence headers are consistent, and the split between core NATS (at-most-once) and JetStream (at-least-once) semantics is clearly explained. Below is my feedback, ordered by severity.

🔴 Blocking / likely bugs

1. NatsStreamMessageConsumer.ReceiveAsync calls MoveNextAsync() on every invocation, even while a prior read is still pending — concurrent enumeration + lost messages.

var next = _enumerator.MoveNextAsync();   // called unconditionally
if (next.IsCompleted) { return [_enumerator.Current.ToMessage()]; }
_pendingMoveNext ??= next.AsTask();       // old pending kept, `next` discarded

The normal message-pump path calls Receive(timeout) repeatedly and gets empty results while idle. On the first timeout, _pendingMoveNext is retained (by design, per the comment). But the next ReceiveAsync call invokes _enumerator.MoveNextAsync() again while the previous one is still in flight. IAsyncEnumerator forbids overlapping MoveNextAsync calls — the NATS ConsumeAsync iterator will throw or corrupt state. Even if it didn't throw, the second MoveNextAsync result is dropped (??= keeps the old task), so a delivered message can be silently consumed and lost. This effectively breaks steady-state JetStream consumption. The guard needs to be "only call MoveNextAsync when there is no pending read":

if (_pendingMoveNext is null)
{
    var next = _enumerator.MoveNextAsync();
    if (next.IsCompletedSuccessfully)
        return next.Result ? [_enumerator.Current.ToMessage()] : [new Message()];
    _pendingMoveNext = next.AsTask();
}

(Also note next.IsCompleted is true for a faulted or end-of-stream false result too; the current code would wrap a stale Current as a message or swallow the fault. Prefer IsCompletedSuccessfully + checking the boolean result.)

This path has no test coverage (see point 5), which is why it wasn't caught.

2. MessageHeaderExtensions.ToNatsHeaders never propagates Header.Bag; it iterates Header.Baggage instead.

foreach (var keyPair in messageHeader.Baggage)   // <-- Baggage, not Bag
{
    if (keyPair.Value != null) headers[keyPair.Key] = keyPair.Value;
}

Baggage is already serialized above into ce-baggage, so this loop re-emits baggage a second time while the user's custom Header.Bag entries are dropped on send. Every other transport (RocketMQ/Kafka/ASB) sends the Bag, filtering local headers via MessageHeader.IsLocalHeader / BagWithoutLocalHeaders(). This looks like it was meant to be messageHeader.Bag and should filter local headers. The DefaultMessageAssertion doesn't check Bag, so the round-trip gap isn't caught by tests.

🟠 Worth addressing

3. JetStream is not enabled in the CI service container. The nats-ci job runs image: nats:latest with no command, so JetStream is off (the docker-compose-nats.yaml correctly passes -js, but CI doesn't use it). Combined with point 5, none of the JetStream code path runs anywhere in CI. If/when JetStream tests are added they'll need command: ["-js"] on the service.

4. nats:latest is a floating tag in both CI and docker-compose-nats.yaml. The rest of the repo's transports pin infra image versions for reproducibility; a pinned tag (e.g. nats:2.10) would avoid surprise breakage from upstream.

5. Test coverage is core-NATS only. NatsMessageGatewayProvider only wires up NatsMessageProducer / NatsSubscription. There are zero tests for NatsStreamMessageProducer, NatsStreamMessageConsumer, stream creation/validation in the factories, publish-confirmation, or ack/nak/term semantics — i.e. the half of the PR with the most complex logic (and bug #1). Given the generated-test harness already exists, a JetStream provider variant would add a lot of confidence.

6. Receive-side Header.Bag pollution. NatsMsgExtensions.ToMessage copies every NATS header (all the ce-* and brighter-* mapped headers) into Header.Bag unfiltered, plus the BrighterNats object. That mixes transport/CloudEvents headers into the user bag; consider excluding the well-known HeadersName.* keys so the bag only carries genuinely custom headers (mirrors the asymmetry from bug #2).

🟡 Minor / style

• docker-compose-nats.yaml still has the obsolete top-level version: '3.8' key (Compose v2 warns and ignores it).
• NatsMessageGatewayConfiguration has an empty parameterless constructor body ({ }) that can be removed since NatsOpts is initialised inline.
• Core-NATS test flakiness risk: core NATS is fire-and-forget, so a publish that races ahead of server-side subscription registration is lost. The provider shares one connection to mitigate this, but there's no explicit flush/round-trip between subscribe and first publish — worth watching for intermittent CI failures.
• NatsStreamMessageConsumer.RejectAsync uses AckTerminateAsync with a TerminateReason (requires NATS Server 2.10.4+, as the doc notes) — fine, but pinning the image (point 4) makes that guarantee concrete.
• The PR body mentions MessagingSystem.NATS; the enum member is actually Nats (fine, just noting the description).

✅ Nice things

• Thorough XML documentation and remarks explaining transport semantics.
• Publish confirmation on the JetStream producer is correctly gated on the server pub-ack and raised off the caller thread with fault containment.
• Sensible NatsNameSanitizer shared between publication and subscription so a generic publication/subscription pair meets on the same stream.
• Good use of OnMissingChannel (Assume/Validate/Create) consistent with the rest of Brighter.

Overall the core-NATS path looks solid; the JetStream path needs bug #1 fixed and test coverage before I'd be comfortable merging. Nice work on the docs and structure.

🤖 Automated review — verify findings before acting on them.