feat: complete remaining jetstream parity implementation plan

2026-02-23 10:16:16 -05:00
parent c7bbf45c8f
commit f46b331921
59 changed files with 1734 additions and 54 deletions
@@ -0,0 +1,16 @@
+namespace NATS.Server.JetStream.Api.Handlers;
+
+public static class AccountApiHandlers
+{
+    public static JetStreamApiResponse HandleInfo(StreamManager streams, ConsumerManager consumers)
+    {
+        return new JetStreamApiResponse
+        {
+            AccountInfo = new JetStreamAccountInfo
+            {
+                Streams = streams.StreamNames.Count,
+                Consumers = consumers.ConsumerCount,
+            },
+        };
+    }
+}
@@ -0,0 +1,23 @@
+namespace NATS.Server.JetStream.Api.Handlers;
+
+public static class ClusterControlApiHandlers
+{
+    public static JetStreamApiResponse HandleMetaLeaderStepdown(JetStream.Cluster.JetStreamMetaGroup meta)
+    {
+        meta.StepDown();
+        return JetStreamApiResponse.SuccessResponse();
+    }
+
+    public static JetStreamApiResponse HandleStreamLeaderStepdown(string subject, StreamManager streams)
+    {
+        if (!subject.StartsWith(JetStreamApiSubjects.StreamLeaderStepdown, StringComparison.Ordinal))
+            return JetStreamApiResponse.NotFound(subject);
+
+        var stream = subject[JetStreamApiSubjects.StreamLeaderStepdown.Length..].Trim();
+        if (stream.Length == 0)
+            return JetStreamApiResponse.NotFound(subject);
+
+        streams.StepDownStreamLeaderAsync(stream, default).GetAwaiter().GetResult();
+        return JetStreamApiResponse.SuccessResponse();
+    }
+}
@@ -1,3 +1,4 @@
+using System.Text;
 using System.Text.Json;
 using NATS.Server.JetStream.Models;

@@ -5,8 +6,15 @@ namespace NATS.Server.JetStream.Api.Handlers;

 public static class ConsumerApiHandlers
 {
-    private const string CreatePrefix = "$JS.API.CONSUMER.CREATE.";
-    private const string InfoPrefix = "$JS.API.CONSUMER.INFO.";
+    private const string CreatePrefix = JetStreamApiSubjects.ConsumerCreate;
+    private const string InfoPrefix = JetStreamApiSubjects.ConsumerInfo;
+    private const string NamesPrefix = JetStreamApiSubjects.ConsumerNames;
+    private const string ListPrefix = JetStreamApiSubjects.ConsumerList;
+    private const string DeletePrefix = JetStreamApiSubjects.ConsumerDelete;
+    private const string PausePrefix = JetStreamApiSubjects.ConsumerPause;
+    private const string ResetPrefix = JetStreamApiSubjects.ConsumerReset;
+    private const string UnpinPrefix = JetStreamApiSubjects.ConsumerUnpin;
+    private const string NextPrefix = JetStreamApiSubjects.ConsumerNext;

    public static JetStreamApiResponse HandleCreate(string subject, ReadOnlySpan<byte> payload, ConsumerManager consumerManager)
    {
@@ -32,6 +40,104 @@ public static class ConsumerApiHandlers
        return consumerManager.GetInfo(stream, durableName);
    }

+    public static JetStreamApiResponse HandleDelete(string subject, ConsumerManager consumerManager)
+    {
+        var parsed = ParseSubject(subject, DeletePrefix);
+        if (parsed == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var (stream, durableName) = parsed.Value;
+        return consumerManager.Delete(stream, durableName)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandleNames(string subject, ConsumerManager consumerManager)
+    {
+        var stream = ParseStreamSubject(subject, NamesPrefix);
+        if (stream == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return new JetStreamApiResponse
+        {
+            ConsumerNames = consumerManager.ListNames(stream),
+        };
+    }
+
+    public static JetStreamApiResponse HandleList(string subject, ConsumerManager consumerManager)
+    {
+        var stream = ParseStreamSubject(subject, ListPrefix);
+        if (stream == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return new JetStreamApiResponse
+        {
+            ConsumerNames = consumerManager.ListNames(stream),
+        };
+    }
+
+    public static JetStreamApiResponse HandlePause(string subject, ReadOnlySpan<byte> payload, ConsumerManager consumerManager)
+    {
+        var parsed = ParseSubject(subject, PausePrefix);
+        if (parsed == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var (stream, durableName) = parsed.Value;
+        var paused = ParsePause(payload);
+        return consumerManager.Pause(stream, durableName, paused)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandleReset(string subject, ConsumerManager consumerManager)
+    {
+        var parsed = ParseSubject(subject, ResetPrefix);
+        if (parsed == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var (stream, durableName) = parsed.Value;
+        return consumerManager.Reset(stream, durableName)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandleUnpin(string subject, ConsumerManager consumerManager)
+    {
+        var parsed = ParseSubject(subject, UnpinPrefix);
+        if (parsed == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var (stream, durableName) = parsed.Value;
+        return consumerManager.Unpin(stream, durableName)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandleNext(string subject, ReadOnlySpan<byte> payload, ConsumerManager consumerManager, StreamManager streamManager)
+    {
+        var parsed = ParseSubject(subject, NextPrefix);
+        if (parsed == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var (stream, durableName) = parsed.Value;
+        var batch = ParseBatch(payload);
+        var pullBatch = consumerManager.FetchAsync(stream, durableName, batch, streamManager, default).GetAwaiter().GetResult();
+        return new JetStreamApiResponse
+        {
+            PullBatch = new JetStreamPullBatch
+            {
+                Messages = pullBatch.Messages
+                    .Select(m => new JetStreamDirectMessage
+                    {
+                        Sequence = m.Sequence,
+                        Subject = m.Subject,
+                        Payload = Encoding.UTF8.GetString(m.Payload.Span),
+                    })
+                    .ToArray(),
+            },
+        };
+    }
+
    private static (string Stream, string Durable)? ParseSubject(string subject, string prefix)
    {
        if (!subject.StartsWith(prefix, StringComparison.Ordinal))
@@ -76,6 +182,8 @@ public static class ConsumerApiHandlers
                var ackPolicy = ackPolicyEl.GetString();
                if (string.Equals(ackPolicy, "explicit", StringComparison.OrdinalIgnoreCase))
                    config.AckPolicy = AckPolicy.Explicit;
+                else if (string.Equals(ackPolicy, "all", StringComparison.OrdinalIgnoreCase))
+                    config.AckPolicy = AckPolicy.All;
            }

            return config;
@@ -85,4 +193,49 @@ public static class ConsumerApiHandlers
            return new ConsumerConfig();
        }
    }
+
+    private static int ParseBatch(ReadOnlySpan<byte> payload)
+    {
+        if (payload.IsEmpty)
+            return 1;
+
+        try
+        {
+            using var doc = JsonDocument.Parse(payload.ToArray());
+            if (doc.RootElement.TryGetProperty("batch", out var batchEl) && batchEl.TryGetInt32(out var batch))
+                return Math.Max(batch, 1);
+        }
+        catch (JsonException)
+        {
+        }
+
+        return 1;
+    }
+
+    private static bool ParsePause(ReadOnlySpan<byte> payload)
+    {
+        if (payload.IsEmpty)
+            return false;
+
+        try
+        {
+            using var doc = JsonDocument.Parse(payload.ToArray());
+            if (doc.RootElement.TryGetProperty("pause", out var pauseEl))
+                return pauseEl.ValueKind == JsonValueKind.True;
+        }
+        catch (JsonException)
+        {
+        }
+
+        return false;
+    }
+
+    private static string? ParseStreamSubject(string subject, string prefix)
+    {
+        if (!subject.StartsWith(prefix, StringComparison.Ordinal))
+            return null;
+
+        var stream = subject[prefix.Length..].Trim();
+        return stream.Length == 0 ? null : stream;
+    }
 }
@@ -0,0 +1,61 @@
+using System.Text;
+using System.Text.Json;
+
+namespace NATS.Server.JetStream.Api.Handlers;
+
+public static class DirectApiHandlers
+{
+    private const string Prefix = JetStreamApiSubjects.DirectGet;
+
+    public static JetStreamApiResponse HandleGet(string subject, ReadOnlySpan<byte> payload, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, Prefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var sequence = ParseSequence(payload);
+        if (sequence == 0)
+            return JetStreamApiResponse.ErrorResponse(400, "sequence required");
+
+        var message = streamManager.GetMessage(streamName, sequence);
+        if (message == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return new JetStreamApiResponse
+        {
+            DirectMessage = new JetStreamDirectMessage
+            {
+                Sequence = message.Sequence,
+                Subject = message.Subject,
+                Payload = Encoding.UTF8.GetString(message.Payload.Span),
+            },
+        };
+    }
+
+    private static string? ExtractTrailingToken(string subject, string prefix)
+    {
+        if (!subject.StartsWith(prefix, StringComparison.Ordinal))
+            return null;
+
+        var token = subject[prefix.Length..].Trim();
+        return token.Length == 0 ? null : token;
+    }
+
+    private static ulong ParseSequence(ReadOnlySpan<byte> payload)
+    {
+        if (payload.IsEmpty)
+            return 0;
+
+        try
+        {
+            using var doc = JsonDocument.Parse(payload.ToArray());
+            if (doc.RootElement.TryGetProperty("seq", out var seqEl) && seqEl.TryGetUInt64(out var sequence))
+                return sequence;
+        }
+        catch (JsonException)
+        {
+        }
+
+        return 0;
+    }
+}
@@ -1,12 +1,20 @@
 using System.Text.Json;
+using System.Text;
 using NATS.Server.JetStream.Models;

 namespace NATS.Server.JetStream.Api.Handlers;

 public static class StreamApiHandlers
 {
-    private const string CreatePrefix = "$JS.API.STREAM.CREATE.";
-    private const string InfoPrefix = "$JS.API.STREAM.INFO.";
+    private const string CreatePrefix = JetStreamApiSubjects.StreamCreate;
+    private const string InfoPrefix = JetStreamApiSubjects.StreamInfo;
+    private const string UpdatePrefix = JetStreamApiSubjects.StreamUpdate;
+    private const string DeletePrefix = JetStreamApiSubjects.StreamDelete;
+    private const string PurgePrefix = JetStreamApiSubjects.StreamPurge;
+    private const string MessageGetPrefix = JetStreamApiSubjects.StreamMessageGet;
+    private const string MessageDeletePrefix = JetStreamApiSubjects.StreamMessageDelete;
+    private const string SnapshotPrefix = JetStreamApiSubjects.StreamSnapshot;
+    private const string RestorePrefix = JetStreamApiSubjects.StreamRestore;

    public static JetStreamApiResponse HandleCreate(string subject, ReadOnlySpan<byte> payload, StreamManager streamManager)
    {
@@ -33,6 +41,131 @@ public static class StreamApiHandlers
        return streamManager.GetInfo(streamName);
    }

+    public static JetStreamApiResponse HandleUpdate(string subject, ReadOnlySpan<byte> payload, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, UpdatePrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var config = ParseConfig(payload);
+        if (string.IsNullOrWhiteSpace(config.Name))
+            config.Name = streamName;
+
+        if (config.Subjects.Count == 0)
+            config.Subjects.Add(streamName.ToLowerInvariant() + ".>");
+
+        return streamManager.CreateOrUpdate(config);
+    }
+
+    public static JetStreamApiResponse HandleDelete(string subject, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, DeletePrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return streamManager.Delete(streamName)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandlePurge(string subject, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, PurgePrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return streamManager.Purge(streamName)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandleNames(StreamManager streamManager)
+    {
+        return new JetStreamApiResponse
+        {
+            StreamNames = streamManager.ListNames(),
+        };
+    }
+
+    public static JetStreamApiResponse HandleList(StreamManager streamManager)
+    {
+        return HandleNames(streamManager);
+    }
+
+    public static JetStreamApiResponse HandleMessageGet(string subject, ReadOnlySpan<byte> payload, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, MessageGetPrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var sequence = ParseSequence(payload);
+        if (sequence == 0)
+            return JetStreamApiResponse.ErrorResponse(400, "sequence required");
+
+        var message = streamManager.GetMessage(streamName, sequence);
+        if (message == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return new JetStreamApiResponse
+        {
+            StreamMessage = new JetStreamStreamMessage
+            {
+                Sequence = message.Sequence,
+                Subject = message.Subject,
+                Payload = Encoding.UTF8.GetString(message.Payload.Span),
+            },
+        };
+    }
+
+    public static JetStreamApiResponse HandleMessageDelete(string subject, ReadOnlySpan<byte> payload, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, MessageDeletePrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var sequence = ParseSequence(payload);
+        if (sequence == 0)
+            return JetStreamApiResponse.ErrorResponse(400, "sequence required");
+
+        return streamManager.DeleteMessage(streamName, sequence)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
+    public static JetStreamApiResponse HandleSnapshot(string subject, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, SnapshotPrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var snapshot = streamManager.CreateSnapshot(streamName);
+        if (snapshot == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        return new JetStreamApiResponse
+        {
+            Snapshot = new JetStreamSnapshot
+            {
+                Payload = Convert.ToBase64String(snapshot),
+            },
+        };
+    }
+
+    public static JetStreamApiResponse HandleRestore(string subject, ReadOnlySpan<byte> payload, StreamManager streamManager)
+    {
+        var streamName = ExtractTrailingToken(subject, RestorePrefix);
+        if (streamName == null)
+            return JetStreamApiResponse.NotFound(subject);
+
+        var snapshotBytes = ParseRestorePayload(payload);
+        if (snapshotBytes == null)
+            return JetStreamApiResponse.ErrorResponse(400, "snapshot payload required");
+
+        return streamManager.RestoreSnapshot(streamName, snapshotBytes)
+            ? JetStreamApiResponse.SuccessResponse()
+            : JetStreamApiResponse.NotFound(subject);
+    }
+
    private static string? ExtractTrailingToken(string subject, string prefix)
    {
        if (!subject.StartsWith(prefix, StringComparison.Ordinal))
@@ -88,4 +221,56 @@ public static class StreamApiHandlers
            return new StreamConfig();
        }
    }
+
+    private static ulong ParseSequence(ReadOnlySpan<byte> payload)
+    {
+        if (payload.IsEmpty)
+            return 0;
+
+        try
+        {
+            using var doc = JsonDocument.Parse(payload.ToArray());
+            if (doc.RootElement.TryGetProperty("seq", out var seqEl) && seqEl.TryGetUInt64(out var sequence))
+                return sequence;
+        }
+        catch (JsonException)
+        {
+        }
+
+        return 0;
+    }
+
+    private static byte[]? ParseRestorePayload(ReadOnlySpan<byte> payload)
+    {
+        if (payload.IsEmpty)
+            return null;
+
+        var raw = Encoding.UTF8.GetString(payload).Trim();
+        if (raw.Length == 0)
+            return null;
+
+        try
+        {
+            return Convert.FromBase64String(raw);
+        }
+        catch (FormatException)
+        {
+        }
+
+        try
+        {
+            using var doc = JsonDocument.Parse(payload.ToArray());
+            if (doc.RootElement.TryGetProperty("payload", out var payloadEl))
+            {
+                var base64 = payloadEl.GetString();
+                if (!string.IsNullOrWhiteSpace(base64))
+                    return Convert.FromBase64String(base64);
+            }
+        }
+        catch (JsonException)
+        {
+        }
+
+        return null;
+    }
 }