Add JetStream metadata to msg parsed from reply string (#139)

* Add seq and datetime to msg parsed from reply string

* Map NatsJSMsg Seq, DateTime to the underlying message

* Correct formatting

* Implement Msg DateTime and Sequence only in JetStream

* Parse reply based on official JSAck ADR

---------

Co-authored-by: Simon Hoss <[email protected]>

.editorconfig

@@ -13,6 +13,7 @@ generated_code = true
 
 [*.cs]
 indent_size = 4
+max_line_length = 300
 
 # changes from VS2019 defaults
 csharp_style_namespace_declarations = file_scoped

src/NATS.Client.JetStream/Internal/ReplyToDateTimeAndSeq.cs

@@ -0,0 +1,74 @@
+namespace NATS.Client.JetStream.Internal;
+
+internal static class ReplyToDateTimeAndSeq
+{
+    private const int V1TokenCounts = 9;
+    private const int V2TokenCounts = 12;
+
+    private const int AckDomainTokenPos = 2;
+    private const int AckAccHashTokenPos = 3;
+    private const int AckStreamTokenPos = 4;
+    private const int AckConsumerTokenPos = 5;
+    private const int AckNumDeliveredTokenPos = 6;
+    private const int AckStreamSeqTokenPos = 7;
+    private const int AckConsumerSeqTokenPos = 8;
+    private const int AckTimestampSeqTokenPos = 9;
+    private const int AckNumPendingTokenPos = 10;
+
+    internal static NatsJSMsgMetadata? Parse(string? reply)
+    {
+        if (reply == null)
+        {
+            return null;
+        }
+
+        var originalTokens = reply.Split(".").AsSpan();
+        var tokensLength = originalTokens.Length;
+
+        // Parsed based on https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-15.md#jsack
+
+        // If lower than 9 or more than 9 but less than 11, report an error
+        if (tokensLength is < V1TokenCounts or > V1TokenCounts and < V2TokenCounts - 1)
+        {
+            return null;
+        }
+
+        if (originalTokens[0] != "$JS" || originalTokens[1] != "ACK")
+        {
+            return null;
+        }
+
+        var tokens = new string[V2TokenCounts].AsSpan();
+
+        if (tokensLength == V1TokenCounts)
+        {
+            originalTokens[..2].CopyTo(tokens);
+            originalTokens[2..].CopyTo(tokens[4..]);
+
+            tokens[AckDomainTokenPos] = string.Empty;
+            tokens[AckAccHashTokenPos] = string.Empty;
+        }
+        else
+        {
+            tokens = originalTokens;
+
+            if (tokens[AckDomainTokenPos] == "_")
+            {
+                tokens[AckDomainTokenPos] = string.Empty;
+            }
+        }
+
+        var timestamp = long.Parse(tokens[AckTimestampSeqTokenPos]);
+        var offset = DateTimeOffset.FromUnixTimeMilliseconds(timestamp / 1000000);
+        var dateTime = new DateTimeOffset(offset.Ticks, TimeSpan.Zero);
+
+        return new NatsJSMsgMetadata(
+            new NatsJSSequencePair(ulong.Parse(tokens[AckStreamSeqTokenPos]), ulong.Parse(tokens[AckConsumerSeqTokenPos])),
+            ulong.Parse(tokens[AckNumDeliveredTokenPos]),
+            ulong.Parse(tokens[AckNumPendingTokenPos]),
+            dateTime,
+            tokens[AckStreamTokenPos],
+            tokens[AckConsumerTokenPos],
+            tokens[AckDomainTokenPos]);
+    }
+}

src/NATS.Client.JetStream/NatsJSMsg.cs

@@ -12,11 +12,13 @@ public readonly struct NatsJSMsg<T>
 {
     private readonly NatsJSContext _context;
     private readonly NatsMsg<T> _msg;
+    private readonly Lazy<NatsJSMsgMetadata?> _replyToDateTimeAndSeq;
 
     internal NatsJSMsg(NatsMsg<T> msg, NatsJSContext context)
     {
         _msg = msg;
         _context = context;
+        _replyToDateTimeAndSeq = new Lazy<NatsJSMsgMetadata?>(() => ReplyToDateTimeAndSeq.Parse(msg.ReplyTo));
     }
 
     /// <summary>
@@ -50,6 +52,11 @@ internal NatsJSMsg(NatsMsg<T> msg, NatsJSContext context)
     /// </summary>
     public INatsConnection? Connection => _msg.Connection;
 
+    /// <summary>
+    /// Additional metadata about the message.
+    /// </summary>
+    public NatsJSMsgMetadata? Metadata => _replyToDateTimeAndSeq.Value;
+
     /// <summary>
     /// Acknowledges the message was completely handled.
     /// </summary>
@@ -102,10 +109,7 @@ private ValueTask SendAckAsync(ReadOnlySequence<byte> payload, AckOpts opts = de
 
         return _msg.ReplyAsync(
             payload: payload,
-            opts: new NatsPubOpts
-            {
-                WaitUntilSent = opts.WaitUntilSent ?? _context.Opts.AckOpts.WaitUntilSent,
-            },
+            opts: new NatsPubOpts {WaitUntilSent = opts.WaitUntilSent ?? _context.Opts.AckOpts.WaitUntilSent,},
             cancellationToken: cancellationToken);
     }
 }

src/NATS.Client.JetStream/NatsJSMsgMetadata.cs

@@ -0,0 +1,38 @@
+namespace NATS.Client.JetStream;
+
+/// <summary>
+/// Additional metadata about the message.
+/// </summary>
+/// <param name="Sequence">
+/// The sequence pair for the message.
+/// </param>
+/// <param name="NumDelivered">
+/// The number of times the message was delivered.
+/// </param>
+/// <param name="NumPending">
+/// The number of messages pending for the consumer.
+/// </param>
+/// <param name="Timestamp">
+/// The timestamp of the message.
+/// </param>
+/// <param name="Stream">
+/// The stream the message was sent to.
+/// </param>
+/// <param name="Consumer">
+/// The consumer the message was sent to.
+/// </param>
+/// <param name="Domain">
+/// The domain the message was sent to.
+/// </param>
+public readonly record struct NatsJSMsgMetadata(NatsJSSequencePair Sequence, ulong NumDelivered, ulong NumPending, DateTimeOffset Timestamp, string Stream, string Consumer, string Domain);
+
+/// <summary>
+/// The sequence pair for the message.
+/// </summary>
+/// <param name="Stream">
+/// The stream sequence number.
+/// </param>
+/// <param name="Consumer">
+/// The consumer sequence number.
+/// </param>
+public readonly record struct NatsJSSequencePair(ulong Stream, ulong Consumer);

tests/NATS.Client.JetStream.Tests/Internal/ReplyToDateTimeAndSeqTest.cs

@@ -0,0 +1,60 @@
+using NATS.Client.JetStream.Internal;
+
+namespace NATS.Client.JetStream.Tests.Internal;
+
+public class ReplyToDateTimeAndSeqTest
+{
+    [Fact]
+    public void ShouldParseV1ReplyToDateTimeAndSeq()
+    {
+        var natsJSMsgMetadata = ReplyToDateTimeAndSeq.Parse("$JS.ACK.UnitTest.GetEvents_0.1.100.1.1696023331771188000.0");
+
+        natsJSMsgMetadata!.Value.Timestamp.ToString("O").Should().Be("2023-09-29T21:35:31.7710000+00:00");
+        natsJSMsgMetadata.Value.Sequence.Stream.Should().Be(100);
+        natsJSMsgMetadata.Value.Sequence.Consumer.Should().Be(1);
+        natsJSMsgMetadata.Value.NumDelivered.Should().Be(1);
+        natsJSMsgMetadata.Value.NumPending.Should().Be(0);
+        natsJSMsgMetadata.Value.Stream.Should().Be("UnitTest");
+        natsJSMsgMetadata.Value.Consumer.Should().Be("GetEvents_0");
+        natsJSMsgMetadata.Value.Domain.Should().BeEmpty();
+    }
+
+    [Fact]
+    public void ShouldParseV2ReplyToDateTimeAndSeq()
+    {
+        var natsJSMsgMetadata = ReplyToDateTimeAndSeq.Parse("$JS.ACK.MyDomain.1234.UnitTest.GetEvents_0.1.100.1.1696023331771188000.0");
+
+        natsJSMsgMetadata!.Value.Timestamp.ToString("O").Should().Be("2023-09-29T21:35:31.7710000+00:00");
+        natsJSMsgMetadata.Value.Sequence.Stream.Should().Be(100);
+        natsJSMsgMetadata.Value.Sequence.Consumer.Should().Be(1);
+        natsJSMsgMetadata.Value.NumDelivered.Should().Be(1);
+        natsJSMsgMetadata.Value.NumPending.Should().Be(0);
+        natsJSMsgMetadata.Value.Stream.Should().Be("UnitTest");
+        natsJSMsgMetadata.Value.Consumer.Should().Be("GetEvents_0");
+        natsJSMsgMetadata.Value.Domain.Should().Be("MyDomain");
+    }
+
+    [Fact]
+    public void ShouldSetNullForReturnWhenReplyToIsNull()
+    {
+        var natsJSMsgMetadata = ReplyToDateTimeAndSeq.Parse(null);
+
+        natsJSMsgMetadata.Should().BeNull();
+    }
+
+    [Fact]
+    public void ShouldSetNullWhenReplyToIsASimpleReqResponse()
+    {
+        var natsJSMsgMetadata = ReplyToDateTimeAndSeq.Parse("_INBOX.1");
+
+        natsJSMsgMetadata.Should().BeNull();
+    }
+
+    [Fact]
+    public void ShouldSetNullWhenDoesNotStartWithJsAck()
+    {
+        var natsJSMsgMetadata = ReplyToDateTimeAndSeq.Parse("1.2.3.4.5.6.7.8.9");
+
+        natsJSMsgMetadata.Should().BeNull();
+    }
+}