using System.Text;
using Shouldly;
using ZB.MOM.NatsNet.Server;

namespace ZB.MOM.NatsNet.Server.Tests.JetStream;

public sealed class NatsConsumerDispatchTests
{
    [Fact]
    public void ProcessWaiting_EndOfStream_ShouldExpireNoWaitRequests()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.ProcessNextMsgRequest("_INBOX.a", Encoding.UTF8.GetBytes("{\"batch\":1,\"no_wait\":true}")).ShouldBeTrue();

        var result = consumer.ProcessWaiting(endOfStream: true);

        result.Expired.ShouldBe(1);
        result.Waiting.ShouldBe(0);
        consumer.CheckWaitingForInterest().ShouldBeFalse();
    }

    [Fact]
    public void HbTimer_HeartbeatConfigured_ShouldReturnTimer()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8, heartbeat: TimeSpan.FromMilliseconds(50));
        var (duration, timer) = consumer.HbTimer();

        duration.ShouldBe(TimeSpan.FromMilliseconds(50));
        timer.ShouldNotBeNull();
        timer!.Dispose();
    }

    [Fact]
    public void CheckAckFloor_WithPendingEntries_ShouldAdvanceFloor()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.ApplyState(new ConsumerState
        {
            Delivered = new SequencePair { Consumer = 20, Stream = 20 },
            AckFloor = new SequencePair { Consumer = 0, Stream = 0 },
            Pending = new Dictionary<ulong, Pending>
            {
                [10] = new Pending { Sequence = 3, Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() },
            },
        });

        consumer.CheckAckFloor();
        var state = consumer.ReadStoredState();

        state.AckFloor.Stream.ShouldBe(9UL);
        state.AckFloor.Consumer.ShouldBe(2UL);
    }

    [Fact]
    public void ProcessInboundAcks_QueuedAck_ShouldAdvanceAckFloor()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.PushAck("$JS.ACK.2.8.1", "reply", 0, Encoding.ASCII.GetBytes("+ACK"));

        var processed = consumer.ProcessInboundAcks(CancellationToken.None);
        var state = consumer.GetConsumerState();

        processed.ShouldBe(1);
        state.AckFloor.Stream.ShouldBe(8UL);
    }

    [Fact]
    public void ProcessInboundNextMsgReqs_QueuedRequest_ShouldPopulateWaitingQueue()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.ProcessNextMsgReq("_INBOX.next", Encoding.UTF8.GetBytes("{\"batch\":2}"));

        var processed = consumer.ProcessInboundNextMsgReqs(CancellationToken.None);
        var pending = consumer.PendingRequests();

        processed.ShouldBe(1);
        pending.ShouldContainKey("_INBOX.next");
        pending["_INBOX.next"].N.ShouldBe(2);
    }

    [Fact]
    public void PendingCounters_AndAckReply_ShouldTrackValues()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.ApplyState(new ConsumerState
        {
            Delivered = new SequencePair { Consumer = 10, Stream = 30 },
            AckFloor = new SequencePair { Consumer = 7, Stream = 22 },
        });

        var (pending, error) = consumer.CheckNumPending();
        error.ShouldBeNull();
        pending.ShouldBe(8UL);
        consumer.NumPending().ShouldBe(8UL);
        consumer.CheckNumPendingOnEOF();

        consumer.SetMaxPendingBytes(256);
        var ackReply = consumer.AckReply(30, 11, 1, 12345, pending);
        ackReply.ShouldContain("$JS.ACK.1.30.11.12345.8");
    }

    [Fact]
    public void SendIdleHeartbeat_ShouldReturnFormattedHeartbeat()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.ApplyState(new ConsumerState
        {
            Delivered = new SequencePair { Consumer = 5, Stream = 9 },
            AckFloor = new SequencePair { Consumer = 4, Stream = 8 },
        });

        var heartbeat = consumer.SendIdleHeartbeat("$JS.HEARTBEAT");

        heartbeat.ShouldContain("100 Idle Heartbeat");
        heartbeat.ShouldContain("Nats-Last-Consumer: 5");
        heartbeat.ShouldContain("Nats-Last-Stream: 9");
    }

    [Fact]
    public void LoopAndGatherMsgs_WithPendingEntries_ShouldDeliverMessages()
    {
        var consumer = CreatePullConsumer(maxWaiting: 8);
        consumer.ApplyState(new ConsumerState
        {
            Delivered = new SequencePair { Consumer = 0, Stream = 0 },
            AckFloor = new SequencePair { Consumer = 0, Stream = 0 },
            Pending = new Dictionary<ulong, Pending>
            {
                [1] = new Pending { Sequence = 1, Timestamp = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() },
            },
        });

        consumer.ProcessNextMsgRequest("_INBOX.loop", Encoding.UTF8.GetBytes("{\"batch\":1}")).ShouldBeTrue();
        var delivered = consumer.LoopAndGatherMsgs(4, CancellationToken.None);

        delivered.ShouldBeGreaterThan(0);
        consumer.GetConsumerState().Delivered.Stream.ShouldBeGreaterThanOrEqualTo(1UL);
    }

    private static NatsConsumer CreatePullConsumer(int maxWaiting, TimeSpan? heartbeat = null)
    {
        var stream = NatsStream.Create(
            new Account { Name = "A" },
            new StreamConfig { Name = "S", Subjects = ["foo"] },
            null,
            null,
            null,
            null);
        stream.ShouldNotBeNull();

        var config = new ConsumerConfig
        {
            Durable = "D",
            MaxWaiting = maxWaiting,
            Heartbeat = heartbeat ?? TimeSpan.Zero,
        };
        var consumer = NatsConsumer.Create(stream!, config, ConsumerAction.CreateOrUpdate, null);
        consumer.ShouldNotBeNull();
        return consumer!;
    }
}