natsdotnet/tests/NATS.E2E.Tests/JetStreamTests.cs

using NATS.Client.Core;
using NATS.Client.JetStream;
using NATS.Client.JetStream.Models;
using NATS.E2E.Tests.Infrastructure;

namespace NATS.E2E.Tests;

[Collection("E2E-JetStream")]
public class JetStreamTests(JetStreamServerFixture fixture)
{
    // -------------------------------------------------------------------------
    // Test 1 — Create a stream and verify its reported info matches config
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_CreateAndInfo()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        var stream = await js.CreateStreamAsync(
            new StreamConfig("E2E_CREATE", ["js.create.>"]),
            cts.Token);

        stream.Info.Config.Name.ShouldBe("E2E_CREATE");
        stream.Info.Config.Subjects.ShouldNotBeNull();
        stream.Info.Config.Subjects.ShouldContain("js.create.>");
    }

    // -------------------------------------------------------------------------
    // Test 2 — List streams and verify all created streams appear
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_ListAndNames()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(new StreamConfig("E2E_LIST_A", ["js.list.a.>"]), cts.Token);
        await js.CreateStreamAsync(new StreamConfig("E2E_LIST_B", ["js.list.b.>"]), cts.Token);
        await js.CreateStreamAsync(new StreamConfig("E2E_LIST_C", ["js.list.c.>"]), cts.Token);

        var names = new List<string>();
        await foreach (var stream in js.ListStreamsAsync(cancellationToken: cts.Token))
        {
            var name = stream.Info.Config.Name;
            if (name is not null)
                names.Add(name);
        }

        names.ShouldContain("E2E_LIST_A");
        names.ShouldContain("E2E_LIST_B");
        names.ShouldContain("E2E_LIST_C");
    }

    // -------------------------------------------------------------------------
    // Test 3 — Delete a stream and verify it is gone
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_Delete()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(new StreamConfig("E2E_DEL", ["js.del.>"]), cts.Token);
        await js.DeleteStreamAsync("E2E_DEL", cts.Token);

        await Should.ThrowAsync<NatsJSApiException>(async () =>
            await js.GetStreamAsync("E2E_DEL", cancellationToken: cts.Token));
    }

    // -------------------------------------------------------------------------
    // Test 4 — Publish messages and verify stream state reflects them
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_PublishAndGet()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        // Use unique stream name to avoid contamination from parallel tests
        var streamName = $"E2E_PUB_{Random.Shared.Next(100000)}";
        await js.CreateStreamAsync(new StreamConfig(streamName, [$"js.pub.{streamName}.>"]), cts.Token);

        await js.PublishAsync($"js.pub.{streamName}.one", "msg1", cancellationToken: cts.Token);
        await js.PublishAsync($"js.pub.{streamName}.two", "msg2", cancellationToken: cts.Token);
        await js.PublishAsync($"js.pub.{streamName}.three", "msg3", cancellationToken: cts.Token);

        var stream = await js.GetStreamAsync(streamName, cancellationToken: cts.Token);
        stream.Info.State.Messages.ShouldBe(3L);
    }

    // -------------------------------------------------------------------------
    // Test 5 — Purge a stream and verify message count drops to zero
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_Purge()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(new StreamConfig("E2E_PURGE", ["js.purge.>"]), cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.purge.msg{i}", $"data{i}", cancellationToken: cts.Token);

        var stream = await js.GetStreamAsync("E2E_PURGE", cancellationToken: cts.Token);
        await stream.PurgeAsync(new StreamPurgeRequest(), cts.Token);

        var refreshed = await js.GetStreamAsync("E2E_PURGE", cancellationToken: cts.Token);
        refreshed.Info.State.Messages.ShouldBe(0L);
    }

    // -------------------------------------------------------------------------
    // Test 6 — Create a durable pull consumer and fetch messages
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_CreatePullAndConsume()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(new StreamConfig("E2E_PULL", ["js.pull.>"]), cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.pull.msg{i}", $"payload{i}", cancellationToken: cts.Token);

        await js.CreateOrUpdateConsumerAsync("E2E_PULL",
            new ConsumerConfig { Name = "pull-consumer", AckPolicy = ConsumerConfigAckPolicy.Explicit },
            cts.Token);

        var consumer = await js.GetConsumerAsync("E2E_PULL", "pull-consumer", cts.Token);

        var received = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 5 }, cancellationToken: cts.Token))
        {
            received.Add(msg.Data);
            await msg.AckAsync(cancellationToken: cts.Token);
        }

        received.Count.ShouldBe(5);
    }

    // -------------------------------------------------------------------------
    // Test 7 — Explicit ack: fetching after ack yields no further messages
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_AckExplicit()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(new StreamConfig("E2E_ACK", ["js.ack.>"]), cts.Token);
        await js.PublishAsync("js.ack.one", "hello", cancellationToken: cts.Token);

        await js.CreateOrUpdateConsumerAsync("E2E_ACK",
            new ConsumerConfig { Name = "ack-consumer", AckPolicy = ConsumerConfigAckPolicy.Explicit },
            cts.Token);

        var consumer = await js.GetConsumerAsync("E2E_ACK", "ack-consumer", cts.Token);

        // Fetch and ack the single message
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 1 }, cancellationToken: cts.Token))
            await msg.AckAsync(cancellationToken: cts.Token);

        // Second fetch should return nothing
        var second = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 1, Expires = TimeSpan.FromSeconds(1) }, cancellationToken: cts.Token))
            second.Add(msg.Data);

        second.Count.ShouldBe(0);
    }

    // -------------------------------------------------------------------------
    // Test 8 — List consumers, delete one, verify count drops
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_ListAndDelete()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(new StreamConfig("E2E_CONS_LIST", ["js.conslist.>"]), cts.Token);

        await js.CreateOrUpdateConsumerAsync("E2E_CONS_LIST",
            new ConsumerConfig { Name = "cons-one", AckPolicy = ConsumerConfigAckPolicy.None },
            cts.Token);
        await js.CreateOrUpdateConsumerAsync("E2E_CONS_LIST",
            new ConsumerConfig { Name = "cons-two", AckPolicy = ConsumerConfigAckPolicy.None },
            cts.Token);

        var beforeNames = new List<string>();
        await foreach (var c in js.ListConsumersAsync("E2E_CONS_LIST", cts.Token))
        {
            var name = c.Info.Name;
            if (name is not null)
                beforeNames.Add(name);
        }

        beforeNames.Count.ShouldBe(2);

        await js.DeleteConsumerAsync("E2E_CONS_LIST", "cons-one", cts.Token);

        var afterNames = new List<string>();
        await foreach (var c in js.ListConsumersAsync("E2E_CONS_LIST", cts.Token))
        {
            var name = c.Info.Name;
            if (name is not null)
                afterNames.Add(name);
        }

        afterNames.Count.ShouldBe(1);
        afterNames.ShouldContain("cons-two");
    }

    // -------------------------------------------------------------------------
    // Test 9 — MaxMsgs retention evicts oldest messages when limit is reached
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Retention_LimitsMaxMessages()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));

        await js.CreateStreamAsync(
            new StreamConfig("E2E_MAXMSGS", ["js.maxmsgs.>"])
            {
                MaxMsgs = 10,
            },
            cts.Token);

        for (var i = 0; i < 15; i++)
            await js.PublishAsync($"js.maxmsgs.{i}", $"val{i}", cancellationToken: cts.Token);

        var stream = await js.GetStreamAsync("E2E_MAXMSGS", cancellationToken: cts.Token);
        stream.Info.State.Messages.ShouldBe(10L);
    }

    // -------------------------------------------------------------------------
    // Test 10 — MaxAge retention expires messages after the configured window
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Retention_MaxAge()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));

        await js.CreateStreamAsync(
            new StreamConfig("E2E_MAXAGE", ["js.maxage.>"])
            {
                MaxAge = TimeSpan.FromSeconds(2),
            },
            cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.maxage.{i}", $"val{i}", cancellationToken: cts.Token);

        var before = await js.GetStreamAsync("E2E_MAXAGE", cancellationToken: cts.Token);
        before.Info.State.Messages.ShouldBe(5L);

        // Poll until MaxAge expiry drops the message count to zero
        INatsJSStream after;
        do
        {
            after = await js.GetStreamAsync("E2E_MAXAGE", cancellationToken: cts.Token);
            if (after.Info.State.Messages == 0L) break;
            await Task.Yield();
        } while (!cts.IsCancellationRequested);

        after.Info.State.Messages.ShouldBe(0L);
    }

    // -------------------------------------------------------------------------
    // Test 11 — Push consumer: consumer is created with DeliverSubject and queryable
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_PushDelivery()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var streamName = $"E2E_PUSH_{Random.Shared.Next(100000)}";
        var deliverSubject = $"_deliver.{streamName}";
        await js.CreateStreamAsync(new StreamConfig(streamName, [$"js.push.{streamName}.>"]), cts.Token);

        for (var i = 0; i < 3; i++)
            await js.PublishAsync($"js.push.{streamName}.{i}", $"push{i}", cancellationToken: cts.Token);

        await js.CreateOrUpdateConsumerAsync(streamName,
            new ConsumerConfig
            {
                Name = "push-consumer",
                DeliverSubject = deliverSubject,
                AckPolicy = ConsumerConfigAckPolicy.None,
            },
            cts.Token);

        // Verify the push consumer was created and the deliver subject is reflected in consumer info
        var consumer = await js.GetConsumerAsync(streamName, "push-consumer", cts.Token);
        consumer.Info.Config.DeliverSubject.ShouldBe(deliverSubject);
    }

    // -------------------------------------------------------------------------
    // Test 12 — AckPolicy.None: re-fetch yields nothing (messages auto-acked)
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_AckNone()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var streamName = $"E2E_ACKNONE_{Random.Shared.Next(100000)}";
        await js.CreateStreamAsync(new StreamConfig(streamName, [$"js.acknone.{streamName}.>"]), cts.Token);

        for (var i = 0; i < 3; i++)
            await js.PublishAsync($"js.acknone.{streamName}.{i}", $"msg{i}", cancellationToken: cts.Token);

        await js.CreateOrUpdateConsumerAsync(streamName,
            new ConsumerConfig { Name = "acknone-consumer", AckPolicy = ConsumerConfigAckPolicy.None },
            cts.Token);

        var consumer = await js.GetConsumerAsync(streamName, "acknone-consumer", cts.Token);

        // First fetch — consume all 3 without acking
        var first = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 3 }, cancellationToken: cts.Token))
            first.Add(msg.Data);

        first.Count.ShouldBe(3);

        // Second fetch — AckNone means server considers them delivered; nothing left
        var second = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 3, Expires = TimeSpan.FromSeconds(1) }, cancellationToken: cts.Token))
            second.Add(msg.Data);

        second.Count.ShouldBe(0);
    }

    // -------------------------------------------------------------------------
    // Test 13 — AckPolicy.All: acking last message acks all prior ones
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_AckAll()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var streamName = $"E2E_ACKALL_{Random.Shared.Next(100000)}";
        await js.CreateStreamAsync(new StreamConfig(streamName, [$"js.ackall.{streamName}.>"]), cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.ackall.{streamName}.{i}", $"msg{i}", cancellationToken: cts.Token);

        await js.CreateOrUpdateConsumerAsync(streamName,
            new ConsumerConfig { Name = "ackall-consumer", AckPolicy = ConsumerConfigAckPolicy.All },
            cts.Token);

        var consumer = await js.GetConsumerAsync(streamName, "ackall-consumer", cts.Token);

        // Fetch all 5, only ack the last one
        INatsJSMsg<string>? last = null;
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 5 }, cancellationToken: cts.Token))
            last = msg;

        last.ShouldNotBeNull();
        await last.AckAsync(cancellationToken: cts.Token);

        // Second fetch should return nothing — all prior msgs are acked via AckAll
        var second = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 5, Expires = TimeSpan.FromSeconds(1) }, cancellationToken: cts.Token))
            second.Add(msg.Data);

        second.Count.ShouldBe(0);
    }

    // -------------------------------------------------------------------------
    // Test 14 — Interest retention: consumer created before publish receives all msgs
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Retention_Interest()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var streamName = $"E2E_INTEREST_{Random.Shared.Next(100000)}";
        await js.CreateStreamAsync(
            new StreamConfig(streamName, [$"js.interest.{streamName}.>"])
            {
                Retention = StreamConfigRetention.Interest,
            },
            cts.Token);

        // Create consumer BEFORE publishing so the server tracks interest
        await js.CreateOrUpdateConsumerAsync(streamName,
            new ConsumerConfig { Name = "interest-consumer", AckPolicy = ConsumerConfigAckPolicy.Explicit },
            cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.interest.{streamName}.{i}", $"msg{i}", cancellationToken: cts.Token);

        var consumer = await js.GetConsumerAsync(streamName, "interest-consumer", cts.Token);

        // Verify the Interest-mode consumer receives all 5 messages and can ack them
        var fetched = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 5 }, cancellationToken: cts.Token))
        {
            fetched.Add(msg.Data);
            await msg.AckAsync(cancellationToken: cts.Token);
        }

        fetched.Count.ShouldBe(5);
    }

    // -------------------------------------------------------------------------
    // Test 15 — WorkQueue retention: messages are stored and fetchable by a single consumer
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Retention_WorkQueue()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var streamName = $"E2E_WQ_{Random.Shared.Next(100000)}";
        await js.CreateStreamAsync(
            new StreamConfig(streamName, [$"js.wq.{streamName}.>"])
            {
                Retention = StreamConfigRetention.Workqueue,
            },
            cts.Token);

        await js.CreateOrUpdateConsumerAsync(streamName,
            new ConsumerConfig { Name = "wq-consumer", AckPolicy = ConsumerConfigAckPolicy.Explicit },
            cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.wq.{streamName}.{i}", $"msg{i}", cancellationToken: cts.Token);

        // Verify all 5 messages are stored in the WorkQueue stream
        var stream = await js.GetStreamAsync(streamName, cancellationToken: cts.Token);
        stream.Info.State.Messages.ShouldBe(5L);

        // Verify the consumer can fetch all 5 messages from the WorkQueue stream
        var consumer = await js.GetConsumerAsync(streamName, "wq-consumer", cts.Token);
        var fetched = new List<string?>();
        await foreach (var msg in consumer.FetchAsync<string>(new NatsJSFetchOpts { MaxMsgs = 5 }, cancellationToken: cts.Token))
            fetched.Add(msg.Data);

        fetched.Count.ShouldBe(5);
    }

    // -------------------------------------------------------------------------
    // Test 16 — Ordered consumer: messages arrive in sequence order
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Consumer_Ordered()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

        var streamName = $"E2E_ORDERED_{Random.Shared.Next(100000)}";
        await js.CreateStreamAsync(new StreamConfig(streamName, [$"js.ordered.{streamName}.>"]), cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.ordered.{streamName}.{i}", i, cancellationToken: cts.Token);

        var consumer = await js.CreateOrderedConsumerAsync(streamName, cancellationToken: cts.Token);

        var sequences = new List<ulong>();
        await foreach (var msg in consumer.FetchAsync<int>(new NatsJSFetchOpts { MaxMsgs = 5 }, cancellationToken: cts.Token))
            sequences.Add(msg.Metadata!.Value.Sequence.Stream);

        sequences.Count.ShouldBe(5);
        for (var i = 1; i < sequences.Count; i++)
            sequences[i].ShouldBeGreaterThan(sequences[i - 1]);
    }

    // -------------------------------------------------------------------------
    // Test 17 — Mirror stream: replicates messages from a source stream
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_Mirror()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));

        var suffix = Random.Shared.Next(100000);
        var sourceName = $"E2E_MIRROR_SRC_{suffix}";
        var mirrorName = $"E2E_MIRROR_DST_{suffix}";

        await js.CreateStreamAsync(new StreamConfig(sourceName, [$"js.mirror.{suffix}.>"]), cts.Token);

        // Create mirror BEFORE publishing so the replication coordinator captures all messages
        await js.CreateStreamAsync(
            new StreamConfig(mirrorName, [])
            {
                Mirror = new StreamSource { Name = sourceName },
            },
            cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.mirror.{suffix}.{i}", $"msg{i}", cancellationToken: cts.Token);

        // Poll until replication completes; outer 30s CTS is the deadline
        INatsJSStream mirror;
        do
        {
            mirror = await js.GetStreamAsync(mirrorName, cancellationToken: cts.Token);
            if (mirror.Info.State.Messages == 5L) break;
            await Task.Yield();
        } while (!cts.IsCancellationRequested);

        mirror.Info.State.Messages.ShouldBe(5L);
    }

    // -------------------------------------------------------------------------
    // Test 18 — Source stream: aggregate stream pulls from a source stream
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Stream_Source()
    {
        await using var client = fixture.CreateClient();
        await client.ConnectAsync();
        var js = new NatsJSContext(client);

        using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));

        var suffix = Random.Shared.Next(100000);
        var srcName = $"E2E_SOURCE_SRC_{suffix}";
        var aggName = $"E2E_SOURCE_AGG_{suffix}";

        await js.CreateStreamAsync(new StreamConfig(srcName, [$"js.source.{suffix}.>"]), cts.Token);

        // Create aggregate stream BEFORE publishing so the source coordinator captures all messages
        await js.CreateStreamAsync(
            new StreamConfig(aggName, [])
            {
                Sources = [new StreamSource { Name = srcName }],
            },
            cts.Token);

        for (var i = 0; i < 5; i++)
            await js.PublishAsync($"js.source.{suffix}.{i}", $"msg{i}", cancellationToken: cts.Token);

        // Poll until sourcing completes; outer 30s CTS is the deadline
        INatsJSStream agg;
        do
        {
            agg = await js.GetStreamAsync(aggName, cancellationToken: cts.Token);
            if (agg.Info.State.Messages == 5L) break;
            await Task.Yield();
        } while (!cts.IsCancellationRequested);

        agg.Info.State.Messages.ShouldBe(5L);
    }
}