// Go parity: golang/nats-server/server/jetstream_cluster_1_test.go
// Covers: cluster stream creation, single/multi replica, memory store,
// stream purge, update subjects, delete, max bytes, stream info/list,
// interest retention, work queue retention, mirror/source in cluster.
using System.Text;
using NATS.Server.JetStream;
using NATS.Server.JetStream.Api;
using NATS.Server.JetStream.Cluster;
using NATS.Server.JetStream.Consumers;
using NATS.Server.JetStream.Models;
using NATS.Server.JetStream.Publish;

namespace NATS.Server.Tests.JetStream.Cluster;

/// <summary>
/// Tests covering clustered JetStream stream creation, replication, storage,
/// purge, update, delete, retention policies, and mirror/source in cluster mode.
/// Ported from Go jetstream_cluster_1_test.go.
/// </summary>
public class JetStreamClusterStreamTests
{
    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterSingleReplicaStreams server/jetstream_cluster_1_test.go:223
    // ---------------------------------------------------------------

    [Fact]
    public async Task Single_replica_stream_creation_and_publish_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var resp = await fx.CreateStreamAsync("R1S", ["foo", "bar"], replicas: 1);
        resp.Error.ShouldBeNull();
        resp.StreamInfo.ShouldNotBeNull();
        resp.StreamInfo!.Config.Name.ShouldBe("R1S");

        const int toSend = 10;
        for (var i = 0; i < toSend; i++)
        {
            var ack = await fx.PublishAsync("foo", $"Hello R1 {i}");
            ack.Stream.ShouldBe("R1S");
            ack.Seq.ShouldBe((ulong)(i + 1));
        }

        var info = await fx.GetStreamInfoAsync("R1S");
        info.StreamInfo!.State.Messages.ShouldBe((ulong)toSend);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterMultiReplicaStreamsDefaultFileMem server/jetstream_cluster_1_test.go:355
    // ---------------------------------------------------------------

    [Fact]
    public async Task Multi_replica_stream_defaults_to_memory_store()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var resp = await fx.CreateStreamAsync("MEMTEST", ["mem.>"], replicas: 3);
        resp.Error.ShouldBeNull();
        resp.StreamInfo!.Config.Storage.ShouldBe(StorageType.Memory);

        var backend = fx.GetStoreBackendType("MEMTEST");
        backend.ShouldBe("memory");
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterMemoryStore server/jetstream_cluster_1_test.go:423
    // ---------------------------------------------------------------

    [Fact]
    public async Task Memory_store_replicated_stream_accepts_100_messages()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var resp = await fx.CreateStreamAsync("R3M", ["foo", "bar"], replicas: 3, storage: StorageType.Memory);
        resp.Error.ShouldBeNull();

        const int toSend = 100;
        for (var i = 0; i < toSend; i++)
        {
            var ack = await fx.PublishAsync("foo", "Hello MemoryStore");
            ack.Stream.ShouldBe("R3M");
        }

        var info = await fx.GetStreamInfoAsync("R3M");
        info.StreamInfo!.Config.Name.ShouldBe("R3M");
        info.StreamInfo.State.Messages.ShouldBe((ulong)toSend);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterDelete server/jetstream_cluster_1_test.go:472
    // ---------------------------------------------------------------

    [Fact]
    public async Task Delete_consumer_then_stream_clears_account_info()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("C22", ["foo", "bar", "baz"], replicas: 2);
        await fx.CreateConsumerAsync("C22", "dlc");

        // Delete consumer then stream
        var delConsumer = await fx.RequestAsync($"{JetStreamApiSubjects.ConsumerDelete}C22.dlc", "{}");
        delConsumer.Success.ShouldBeTrue();

        var delStream = await fx.RequestAsync($"{JetStreamApiSubjects.StreamDelete}C22", "{}");
        delStream.Success.ShouldBeTrue();

        // Account info should show zero streams
        var accountInfo = await fx.RequestAsync(JetStreamApiSubjects.Info, "{}");
        accountInfo.AccountInfo.ShouldNotBeNull();
        accountInfo.AccountInfo!.Streams.ShouldBe(0);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamPurge server/jetstream_cluster_1_test.go:522
    // ---------------------------------------------------------------

    [Fact]
    public async Task Stream_purge_clears_all_messages_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 5);

        await fx.CreateStreamAsync("PURGE", ["foo", "bar"], replicas: 3);

        const int toSend = 100;
        for (var i = 0; i < toSend; i++)
            await fx.PublishAsync("foo", "Hello JS Clustering");

        var before = await fx.GetStreamInfoAsync("PURGE");
        before.StreamInfo!.State.Messages.ShouldBe((ulong)toSend);

        var purge = await fx.RequestAsync($"{JetStreamApiSubjects.StreamPurge}PURGE", "{}");
        purge.Success.ShouldBeTrue();

        var after = await fx.GetStreamInfoAsync("PURGE");
        after.StreamInfo!.State.Messages.ShouldBe(0UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamUpdateSubjects server/jetstream_cluster_1_test.go:571
    // ---------------------------------------------------------------

    [Fact]
    public async Task Stream_update_subjects_reflects_new_configuration()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("SUBUPDATE", ["foo", "bar"], replicas: 3);

        // Update subjects to bar, baz
        var update = fx.UpdateStream("SUBUPDATE", ["bar", "baz"], replicas: 3);
        update.Error.ShouldBeNull();
        update.StreamInfo.ShouldNotBeNull();
        update.StreamInfo!.Config.Subjects.ShouldContain("bar");
        update.StreamInfo.Config.Subjects.ShouldContain("baz");
        update.StreamInfo.Config.Subjects.ShouldNotContain("foo");
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamInfoList server/jetstream_cluster_1_test.go:1284
    // ---------------------------------------------------------------

    [Fact]
    public async Task Stream_names_and_list_return_all_streams()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("S1", ["s1.>"], replicas: 3);
        await fx.CreateStreamAsync("S2", ["s2.>"], replicas: 3);
        await fx.CreateStreamAsync("S3", ["s3.>"], replicas: 1);

        var names = await fx.RequestAsync(JetStreamApiSubjects.StreamNames, "{}");
        names.StreamNames.ShouldNotBeNull();
        names.StreamNames!.Count.ShouldBe(3);
        names.StreamNames.ShouldContain("S1");
        names.StreamNames.ShouldContain("S2");
        names.StreamNames.ShouldContain("S3");

        var list = await fx.RequestAsync(JetStreamApiSubjects.StreamList, "{}");
        list.StreamNames.ShouldNotBeNull();
        list.StreamNames!.Count.ShouldBe(3);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterMaxBytesForStream server/jetstream_cluster_1_test.go:1099
    // ---------------------------------------------------------------

    [Fact]
    public async Task Max_bytes_stream_limits_enforced_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "MAXBYTES",
            Subjects = ["mb.>"],
            Replicas = 3,
            MaxBytes = 512,
            Discard = DiscardPolicy.Old,
        };
        var resp = fx.CreateStreamDirect(cfg);
        resp.Error.ShouldBeNull();

        // Publish messages exceeding max bytes; old messages should be discarded
        for (var i = 0; i < 20; i++)
            await fx.PublishAsync("mb.data", new string('X', 64));

        var state = await fx.GetStreamStateAsync("MAXBYTES");
        // Total bytes should not exceed max_bytes by much after enforcement
        ((long)state.Bytes).ShouldBeLessThanOrEqualTo(cfg.MaxBytes + 128);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamPublishWithActiveConsumers server/jetstream_cluster_1_test.go:1132
    // ---------------------------------------------------------------

    [Fact]
    public async Task Publish_with_active_consumer_delivers_messages()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("ACTIVE", ["active.>"], replicas: 3);
        await fx.CreateConsumerAsync("ACTIVE", "durable1", filterSubject: "active.>");

        for (var i = 0; i < 10; i++)
            await fx.PublishAsync("active.event", $"msg-{i}");

        var batch = await fx.FetchAsync("ACTIVE", "durable1", 10);
        batch.Messages.Count.ShouldBe(10);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterDoubleAdd server/jetstream_cluster_1_test.go:1551
    // ---------------------------------------------------------------

    [Fact]
    public async Task Double_add_stream_with_same_config_succeeds()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var first = await fx.CreateStreamAsync("DUP", ["dup.>"], replicas: 3);
        first.Error.ShouldBeNull();

        // Adding the same stream again should succeed (idempotent)
        var second = await fx.CreateStreamAsync("DUP", ["dup.>"], replicas: 3);
        second.Error.ShouldBeNull();
        second.StreamInfo!.Config.Name.ShouldBe("DUP");
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamOverlapSubjects server/jetstream_cluster_1_test.go:1248
    // ---------------------------------------------------------------

    [Fact]
    public async Task Publish_routes_to_correct_stream_among_non_overlapping()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("ALPHA", ["alpha.>"], replicas: 3);
        await fx.CreateStreamAsync("BETA", ["beta.>"], replicas: 3);

        var ack1 = await fx.PublishAsync("alpha.one", "A");
        ack1.Stream.ShouldBe("ALPHA");

        var ack2 = await fx.PublishAsync("beta.one", "B");
        ack2.Stream.ShouldBe("BETA");
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterInterestRetention server/jetstream_cluster_1_test.go:2109
    // ---------------------------------------------------------------

    [Fact]
    public async Task Interest_retention_stream_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "INTEREST",
            Subjects = ["interest.>"],
            Replicas = 3,
            Retention = RetentionPolicy.Interest,
        };
        fx.CreateStreamDirect(cfg);

        for (var i = 0; i < 5; i++)
            await fx.PublishAsync("interest.event", "msg");

        var state = await fx.GetStreamStateAsync("INTEREST");
        state.Messages.ShouldBe(5UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterWorkQueueRetention server/jetstream_cluster_1_test.go:2179
    // ---------------------------------------------------------------

    [Fact]
    public async Task Work_queue_retention_removes_acked_messages_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "WQ",
            Subjects = ["wq.>"],
            Replicas = 2,
            Retention = RetentionPolicy.WorkQueue,
            MaxConsumers = 1,
        };
        fx.CreateStreamDirect(cfg);
        await fx.CreateConsumerAsync("WQ", "worker", filterSubject: "wq.>", ackPolicy: AckPolicy.All);

        await fx.PublishAsync("wq.task", "job-1");

        var stateBefore = await fx.GetStreamStateAsync("WQ");
        stateBefore.Messages.ShouldBe(1UL);

        // Ack all up to sequence 1, triggering work queue cleanup
        fx.AckAll("WQ", "worker", 1);

        // Publish again to trigger runtime retention enforcement
        await fx.PublishAsync("wq.task", "job-2");

        var stateAfter = await fx.GetStreamStateAsync("WQ");
        // After ack, only the new message should remain
        stateAfter.Messages.ShouldBe(1UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterDeleteMsg server/jetstream_cluster_1_test.go:1748
    // ---------------------------------------------------------------

    [Fact]
    public async Task Delete_individual_message_in_clustered_stream()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("DELMSG", ["dm.>"], replicas: 3);

        for (var i = 0; i < 5; i++)
            await fx.PublishAsync("dm.event", $"msg-{i}");

        var before = await fx.GetStreamStateAsync("DELMSG");
        before.Messages.ShouldBe(5UL);

        // Delete message at sequence 3
        var del = await fx.RequestAsync($"{JetStreamApiSubjects.StreamMessageDelete}DELMSG", """{"seq":3}""");
        del.Success.ShouldBeTrue();

        var after = await fx.GetStreamStateAsync("DELMSG");
        after.Messages.ShouldBe(4UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamUpdate server/jetstream_cluster_1_test.go:1433
    // ---------------------------------------------------------------

    [Fact]
    public async Task Stream_update_preserves_existing_messages()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("UPD", ["upd.>"], replicas: 3);

        for (var i = 0; i < 5; i++)
            await fx.PublishAsync("upd.event", $"msg-{i}");

        // Update max_msgs
        var update = fx.UpdateStream("UPD", ["upd.>"], replicas: 3, maxMsgs: 10);
        update.Error.ShouldBeNull();

        var state = await fx.GetStreamStateAsync("UPD");
        state.Messages.ShouldBe(5UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterAccountInfo server/jetstream_cluster_1_test.go:94
    // ---------------------------------------------------------------

    [Fact]
    public async Task Account_info_reports_stream_and_consumer_counts()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("AI1", ["ai1.>"], replicas: 3);
        await fx.CreateStreamAsync("AI2", ["ai2.>"], replicas: 3);
        await fx.CreateConsumerAsync("AI1", "c1");

        var resp = await fx.RequestAsync(JetStreamApiSubjects.Info, "{}");
        resp.AccountInfo.ShouldNotBeNull();
        resp.AccountInfo!.Streams.ShouldBe(2);
        resp.AccountInfo.Consumers.ShouldBe(1);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterExpand server/jetstream_cluster_1_test.go:86
    // ---------------------------------------------------------------

    [Fact]
    public void Cluster_expand_adds_peer_to_meta_group()
    {
        var meta = new JetStreamMetaGroup(2);
        var state = meta.GetState();
        state.ClusterSize.ShouldBe(2);

        // Expanding is modeled by creating a new meta group with more nodes
        var expanded = new JetStreamMetaGroup(3);
        expanded.GetState().ClusterSize.ShouldBe(3);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterMirrorAndSourceWorkQueues server/jetstream_cluster_1_test.go:2233
    // ---------------------------------------------------------------

    [Fact]
    public async Task Mirror_stream_replicates_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        // Create origin stream
        await fx.CreateStreamAsync("ORIGIN", ["origin.>"], replicas: 3);

        // Create mirror stream
        fx.CreateStreamDirect(new StreamConfig
        {
            Name = "MIRROR",
            Subjects = ["mirror.>"],
            Replicas = 3,
            Mirror = "ORIGIN",
        });

        // Publish to origin
        for (var i = 0; i < 5; i++)
            await fx.PublishAsync("origin.event", $"mirrored-{i}");

        // Mirror should have replicated messages
        var mirrorState = await fx.GetStreamStateAsync("MIRROR");
        mirrorState.Messages.ShouldBe(5UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterMirrorAndSourceInterestPolicyStream server/jetstream_cluster_1_test.go:2290
    // ---------------------------------------------------------------

    [Fact]
    public async Task Source_stream_replicates_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        // Create source origin
        await fx.CreateStreamAsync("SRC", ["src.>"], replicas: 3);

        // Create aggregate stream sourcing from SRC
        fx.CreateStreamDirect(new StreamConfig
        {
            Name = "AGG",
            Subjects = ["agg.>"],
            Replicas = 3,
            Sources = [new StreamSourceConfig { Name = "SRC" }],
        });

        // Publish to source
        for (var i = 0; i < 3; i++)
            await fx.PublishAsync("src.event", $"sourced-{i}");

        var aggState = await fx.GetStreamStateAsync("AGG");
        aggState.Messages.ShouldBe(3UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterUserSnapshotAndRestore server/jetstream_cluster_1_test.go:2652
    // ---------------------------------------------------------------

    [Fact]
    public async Task Snapshot_and_restore_preserves_messages_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("SNAP", ["snap.>"], replicas: 3);

        for (var i = 0; i < 10; i++)
            await fx.PublishAsync("snap.event", $"msg-{i}");

        // Create snapshot
        var snapshot = await fx.RequestAsync($"{JetStreamApiSubjects.StreamSnapshot}SNAP", "{}");
        snapshot.Snapshot.ShouldNotBeNull();
        snapshot.Snapshot!.Payload.ShouldNotBeNullOrEmpty();

        // Purge the stream
        await fx.RequestAsync($"{JetStreamApiSubjects.StreamPurge}SNAP", "{}");
        var afterPurge = await fx.GetStreamStateAsync("SNAP");
        afterPurge.Messages.ShouldBe(0UL);

        // Restore from snapshot
        var restore = await fx.RequestAsync($"{JetStreamApiSubjects.StreamRestore}SNAP", snapshot.Snapshot.Payload);
        restore.Success.ShouldBeTrue();

        var afterRestore = await fx.GetStreamStateAsync("SNAP");
        afterRestore.Messages.ShouldBe(10UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamSynchedTimeStamps server/jetstream_cluster_1_test.go:977
    // ---------------------------------------------------------------

    [Fact]
    public async Task Replicated_stream_messages_have_monotonic_sequences()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("SEQ", ["seq.>"], replicas: 3);

        var sequences = new List<ulong>();
        for (var i = 0; i < 20; i++)
        {
            var ack = await fx.PublishAsync("seq.event", $"msg-{i}");
            sequences.Add(ack.Seq);
        }

        // Verify strictly monotonically increasing sequences
        for (var i = 1; i < sequences.Count; i++)
            sequences[i].ShouldBeGreaterThan(sequences[i - 1]);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamLimits server/jetstream_cluster_1_test.go:3248
    // ---------------------------------------------------------------

    [Fact]
    public async Task Max_msgs_limit_enforced_in_clustered_stream()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "LIMITED",
            Subjects = ["limited.>"],
            Replicas = 3,
            MaxMsgs = 5,
        };
        fx.CreateStreamDirect(cfg);

        for (var i = 0; i < 10; i++)
            await fx.PublishAsync("limited.event", $"msg-{i}");

        var state = await fx.GetStreamStateAsync("LIMITED");
        state.Messages.ShouldBeLessThanOrEqualTo(5UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamInterestOnlyPolicy server/jetstream_cluster_1_test.go:3310
    // ---------------------------------------------------------------

    [Fact]
    public async Task Interest_only_policy_stream_stores_messages_without_consumers()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "INTONLY",
            Subjects = ["intonly.>"],
            Replicas = 3,
            Retention = RetentionPolicy.Interest,
        };
        fx.CreateStreamDirect(cfg);

        for (var i = 0; i < 3; i++)
            await fx.PublishAsync("intonly.data", $"msg-{i}");

        // Without consumers, interest retention still stores messages
        // (they are removed only when all consumers have acked)
        var state = await fx.GetStreamStateAsync("INTONLY");
        state.Messages.ShouldBe(3UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterConsumerInfoList server/jetstream_cluster_1_test.go:1349
    // ---------------------------------------------------------------

    [Fact]
    public async Task Consumer_names_and_list_return_all_consumers()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("CLIST", ["clist.>"], replicas: 3);
        await fx.CreateConsumerAsync("CLIST", "c1");
        await fx.CreateConsumerAsync("CLIST", "c2");
        await fx.CreateConsumerAsync("CLIST", "c3");

        var names = await fx.RequestAsync($"{JetStreamApiSubjects.ConsumerNames}CLIST", "{}");
        names.ConsumerNames.ShouldNotBeNull();
        names.ConsumerNames!.Count.ShouldBe(3);
        names.ConsumerNames.ShouldContain("c1");
        names.ConsumerNames.ShouldContain("c2");
        names.ConsumerNames.ShouldContain("c3");
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterDefaultMaxAckPending server/jetstream_cluster_1_test.go:1580
    // ---------------------------------------------------------------

    [Fact]
    public async Task Consumer_default_ack_policy_is_none()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("ACKDEF", ["ackdef.>"], replicas: 3);
        var resp = await fx.CreateConsumerAsync("ACKDEF", "test_consumer");

        resp.ConsumerInfo.ShouldNotBeNull();
        resp.ConsumerInfo!.Config.AckPolicy.ShouldBe(AckPolicy.None);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterExtendedStreamInfo server/jetstream_cluster_1_test.go:1878
    // ---------------------------------------------------------------

    [Fact]
    public async Task Stream_info_returns_config_and_state()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("EXTINFO", ["ext.>"], replicas: 3);

        for (var i = 0; i < 5; i++)
            await fx.PublishAsync("ext.event", $"msg-{i}");

        var info = await fx.GetStreamInfoAsync("EXTINFO");
        info.StreamInfo.ShouldNotBeNull();
        info.StreamInfo!.Config.Name.ShouldBe("EXTINFO");
        info.StreamInfo.Config.Replicas.ShouldBe(3);
        info.StreamInfo.State.Messages.ShouldBe(5UL);
        info.StreamInfo.State.FirstSeq.ShouldBe(1UL);
        info.StreamInfo.State.LastSeq.ShouldBe(5UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterExtendedStreamInfoSingleReplica server/jetstream_cluster_1_test.go:2033
    // ---------------------------------------------------------------

    [Fact]
    public async Task Single_replica_stream_info_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        await fx.CreateStreamAsync("R1INFO", ["r1info.>"], replicas: 1);

        for (var i = 0; i < 3; i++)
            await fx.PublishAsync("r1info.event", $"msg-{i}");

        var info = await fx.GetStreamInfoAsync("R1INFO");
        info.StreamInfo!.Config.Replicas.ShouldBe(1);
        info.StreamInfo.State.Messages.ShouldBe(3UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterMultiReplicaStreams (maxmsgs_per behavior)
    // ---------------------------------------------------------------

    [Fact]
    public async Task Max_msgs_per_subject_enforced_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "PERSUBJ",
            Subjects = ["ps.>"],
            Replicas = 3,
            MaxMsgsPer = 2,
        };
        fx.CreateStreamDirect(cfg);

        // Publish 5 messages to same subject; only 2 should remain
        for (var i = 0; i < 5; i++)
            await fx.PublishAsync("ps.topic", $"msg-{i}");

        var state = await fx.GetStreamStateAsync("PERSUBJ");
        state.Messages.ShouldBeLessThanOrEqualTo(2UL);
    }

    // ---------------------------------------------------------------
    // Go: TestJetStreamClusterStreamExtendedUpdates server/jetstream_cluster_1_test.go:1513
    // ---------------------------------------------------------------

    [Fact]
    public async Task Stream_update_can_change_max_msgs()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "EXTUPD",
            Subjects = ["eu.>"],
            Replicas = 3,
        };
        fx.CreateStreamDirect(cfg);

        for (var i = 0; i < 10; i++)
            await fx.PublishAsync("eu.event", $"msg-{i}");

        // Update to limit max_msgs
        var update = fx.UpdateStream("EXTUPD", ["eu.>"], replicas: 3, maxMsgs: 5);
        update.Error.ShouldBeNull();
        update.StreamInfo!.Config.MaxMsgs.ShouldBe(5);
    }

    // ---------------------------------------------------------------
    // Additional: Sealed stream rejects purge
    // ---------------------------------------------------------------

    [Fact]
    public async Task Sealed_stream_rejects_purge_in_cluster()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "SEALED",
            Subjects = ["sealed.>"],
            Replicas = 3,
            Sealed = true,
        };
        fx.CreateStreamDirect(cfg);

        var purge = await fx.RequestAsync($"{JetStreamApiSubjects.StreamPurge}SEALED", "{}");
        // Sealed streams should not allow purge
        purge.Success.ShouldBeFalse();
    }

    // ---------------------------------------------------------------
    // Additional: DenyDelete stream rejects message delete
    // ---------------------------------------------------------------

    [Fact]
    public async Task DenyDelete_stream_rejects_message_delete()
    {
        await using var fx = await ClusterStreamFixture.StartAsync(nodes: 3);

        var cfg = new StreamConfig
        {
            Name = "NODELDENY",
            Subjects = ["nodel.>"],
            Replicas = 3,
            DenyDelete = true,
        };
        fx.CreateStreamDirect(cfg);

        await fx.PublishAsync("nodel.event", "msg");

        var del = await fx.RequestAsync($"{JetStreamApiSubjects.StreamMessageDelete}NODELDENY", """{"seq":1}""");
        del.Success.ShouldBeFalse();
    }
}

/// <summary>
/// Self-contained fixture for JetStream cluster stream tests. Wires up
/// meta group, stream manager, consumer manager, API router, and publisher.
/// </summary>
internal sealed class ClusterStreamFixture : IAsyncDisposable
{
    private readonly JetStreamMetaGroup _metaGroup;
    private readonly StreamManager _streamManager;
    private readonly ConsumerManager _consumerManager;
    private readonly JetStreamApiRouter _router;
    private readonly JetStreamPublisher _publisher;

    private ClusterStreamFixture(
        JetStreamMetaGroup metaGroup,
        StreamManager streamManager,
        ConsumerManager consumerManager,
        JetStreamApiRouter router,
        JetStreamPublisher publisher)
    {
        _metaGroup = metaGroup;
        _streamManager = streamManager;
        _consumerManager = consumerManager;
        _router = router;
        _publisher = publisher;
    }

    public static Task<ClusterStreamFixture> StartAsync(int nodes)
    {
        var meta = new JetStreamMetaGroup(nodes);
        var consumerManager = new ConsumerManager(meta);
        var streamManager = new StreamManager(meta, consumerManager: consumerManager);
        var router = new JetStreamApiRouter(streamManager, consumerManager, meta);
        var publisher = new JetStreamPublisher(streamManager);
        return Task.FromResult(new ClusterStreamFixture(meta, streamManager, consumerManager, router, publisher));
    }

    public Task<JetStreamApiResponse> CreateStreamAsync(string name, string[] subjects, int replicas, StorageType storage = StorageType.Memory)
    {
        var response = _streamManager.CreateOrUpdate(new StreamConfig
        {
            Name = name,
            Subjects = [.. subjects],
            Replicas = replicas,
            Storage = storage,
        });
        return Task.FromResult(response);
    }

    public JetStreamApiResponse CreateStreamDirect(StreamConfig config)
        => _streamManager.CreateOrUpdate(config);

    public JetStreamApiResponse UpdateStream(string name, string[] subjects, int replicas, int maxMsgs = 0)
    {
        return _streamManager.CreateOrUpdate(new StreamConfig
        {
            Name = name,
            Subjects = [.. subjects],
            Replicas = replicas,
            MaxMsgs = maxMsgs,
        });
    }

    public Task<PubAck> PublishAsync(string subject, string payload)
    {
        if (_publisher.TryCapture(subject, Encoding.UTF8.GetBytes(payload), null, out var ack))
        {
            if (ack.ErrorCode == null && _streamManager.TryGet(ack.Stream, out var handle))
            {
                var stored = handle.Store.LoadAsync(ack.Seq, default).GetAwaiter().GetResult();
                if (stored != null)
                    _consumerManager.OnPublished(ack.Stream, stored);
            }

            return Task.FromResult(ack);
        }

        throw new InvalidOperationException($"Publish to '{subject}' did not match a stream.");
    }

    public Task<JetStreamApiResponse> GetStreamInfoAsync(string name)
        => Task.FromResult(_streamManager.GetInfo(name));

    public Task<ApiStreamState> GetStreamStateAsync(string name)
        => _streamManager.GetStateAsync(name, default).AsTask();

    public string GetStoreBackendType(string name) => _streamManager.GetStoreBackendType(name);

    public Task<JetStreamApiResponse> CreateConsumerAsync(
        string stream,
        string durableName,
        string? filterSubject = null,
        AckPolicy ackPolicy = AckPolicy.None)
    {
        var config = new ConsumerConfig
        {
            DurableName = durableName,
            AckPolicy = ackPolicy,
        };
        if (!string.IsNullOrWhiteSpace(filterSubject))
            config.FilterSubject = filterSubject;

        return Task.FromResult(_consumerManager.CreateOrUpdate(stream, config));
    }

    public Task<PullFetchBatch> FetchAsync(string stream, string durableName, int batch)
        => _consumerManager.FetchAsync(stream, durableName, batch, _streamManager, default).AsTask();

    public void AckAll(string stream, string durableName, ulong sequence)
        => _consumerManager.AckAll(stream, durableName, sequence);

    public Task<JetStreamApiResponse> RequestAsync(string subject, string payload)
        => Task.FromResult(_router.Route(subject, Encoding.UTF8.GetBytes(payload)));

    public ValueTask DisposeAsync() => ValueTask.CompletedTask;
}