// Copyright 2019-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Adapted from server/consumer.go in the NATS server Go source.

namespace ZB.MOM.NatsNet.Server;

/// <summary>
/// Represents a JetStream consumer, managing message delivery, ack tracking, and lifecycle.
/// Mirrors the <c>consumer</c> struct in server/consumer.go.
/// </summary>
internal sealed partial class NatsConsumer : IDisposable
{
    private readonly ReaderWriterLockSlim _mu = new(LockRecursionPolicy.SupportsRecursion);

    public string Name { get; private set; } = string.Empty;
    public string Stream { get; private set; } = string.Empty;
    public ConsumerConfig Config { get; private set; } = new();
    public DateTime Created { get; private set; }

    // Atomic counters — use Interlocked for thread-safe access
    internal long Delivered;
    internal long AckFloor;
    internal long NumAckPending;
    internal long NumRedelivered;

    private bool _closed;
    private bool _isLeader;
    private ulong _leaderTerm;
    private ConsumerState _state = new();
    private NatsStream? _streamRef;
    private ConsumerAssignment? _assignment;
    private DateTime _lostQuorumSent;
    private TimeSpan _deleteThreshold;
    private bool _isPaused;
    private Timer? _deleteTimer;
    private Timer? _gatewayWatchTimer;
    private bool _hasLocalDeliveryInterest;
    private long _rateLimitBitsPerSecond;
    private int _rateLimitBurstBytes;
    private readonly Queue<byte[]> _proposalQueue = new();
    private readonly Dictionary<ulong, string> _ackReplies = new();
    private readonly Dictionary<string, DateTime> _clusterPendingRequests = new(StringComparer.Ordinal);
    private bool _pendingRequestsOk;
    private WaitQueue? _waiting;
    private Queue<NextMsgReq>? _nextMsgReqs;
    private Timer? _pinnedTtlTimer;
    private string _currentPinId = string.Empty;
    private DateTime _pinnedTs;
    private Dictionary<ulong, ulong>? _rdc;
    private long _npc;
    private ulong _npf;
    private int _maxPendingBytesLimit;
    private int _maxPendingBytesThreshold;
    private long _pendingBytes;
    private int _flowControlWindow;
    private int _flowControlSentBytes;
    private string _flowControlReplyId = string.Empty;
    private readonly Queue<ulong> _redeliveryQueue = new();
    private readonly HashSet<ulong> _redeliveryIndex = new();
    private bool _monitorRunning;
    private long _streamPending;
    private Timer? _pendingTimer;

    /// <summary>IRaftNode — stored as object to avoid cross-dependency on Raft session.</summary>
    private object? _node;

    private CancellationTokenSource? _quitCts;

    public NatsConsumer(string stream, ConsumerConfig config, DateTime created)
    {
        Stream = stream;
        Name = (config.Name is { Length: > 0 } name) ? name
             : (config.Durable ?? string.Empty);
        Config = config;
        Created = created;
        _quitCts = new CancellationTokenSource();
        _waiting = WaitQueue.NewWaitQueue(Math.Max(0, config.MaxWaiting));
        _nextMsgReqs = IsPullMode() ? new Queue<NextMsgReq>() : null;
    }

    // -------------------------------------------------------------------------
    // Factory
    // -------------------------------------------------------------------------

    /// <summary>
    /// Creates a new <see cref="NatsConsumer"/> for the given stream.
    /// Returns null if the consumer cannot be created (stub: always throws).
    /// Mirrors <c>newConsumer</c> / <c>consumer.create</c> in server/consumer.go.
    /// </summary>
    public static NatsConsumer? Create(
        NatsStream stream,
        ConsumerConfig cfg,
        ConsumerAction action,
        ConsumerAssignment? sa)
    {
        ArgumentNullException.ThrowIfNull(stream);
        ArgumentNullException.ThrowIfNull(cfg);
        var consumer = new NatsConsumer(stream.Name, cfg, DateTime.UtcNow)
        {
            _streamRef = stream,
            _assignment = sa,
        };
        return consumer;
    }

    // -------------------------------------------------------------------------
    // Lifecycle
    // -------------------------------------------------------------------------

    /// <summary>
    /// Stops processing and tears down goroutines / timers.
    /// Mirrors <c>consumer.stop</c> in server/consumer.go.
    /// </summary>
    public void Stop()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_closed)
                return;
            _closed = true;
            _isLeader = false;
            _quitCts?.Cancel();
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    /// <summary>
    /// Deletes the consumer and all associated state permanently.
    /// Mirrors <c>consumer.delete</c> in server/consumer.go.
    /// </summary>
    public void Delete() => Stop();

    // -------------------------------------------------------------------------
    // Info / State
    // -------------------------------------------------------------------------

    /// <summary>
    /// Returns a snapshot of consumer info including config and delivery state.
    /// Mirrors <c>consumer.info</c> in server/consumer.go.
    /// </summary>
    public ConsumerInfo GetInfo()
    {
        _mu.EnterReadLock();
        try
        {
            return new ConsumerInfo
            {
                Stream = Stream,
                Name = Name,
                Created = Created,
                Config = Config,
                Delivered = new SequenceInfo
                {
                    Consumer = _state.Delivered.Consumer,
                    Stream = _state.Delivered.Stream,
                },
                AckFloor = new SequenceInfo
                {
                    Consumer = _state.AckFloor.Consumer,
                    Stream = _state.AckFloor.Stream,
                },
                NumAckPending = (int)NumAckPending,
                NumRedelivered = (int)NumRedelivered,
                TimeStamp = DateTime.UtcNow,
            };
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    /// <summary>
    /// Returns the current consumer configuration.
    /// Mirrors <c>consumer.config</c> in server/consumer.go.
    /// </summary>
    public ConsumerConfig GetConfig()
    {
        _mu.EnterReadLock();
        try { return Config; }
        finally { _mu.ExitReadLock(); }
    }

    /// <summary>
    /// Applies an updated configuration to the consumer.
    /// Mirrors <c>consumer.update</c> in server/consumer.go.
    /// </summary>
    public void UpdateConfig(ConsumerConfig config)
    {
        ArgumentNullException.ThrowIfNull(config);
        _mu.EnterWriteLock();
        try { Config = config; }
        finally { _mu.ExitWriteLock(); }
    }

    /// <summary>
    /// Returns the current durable consumer state (delivered, ack_floor, pending, redelivered).
    /// Mirrors <c>consumer.state</c> in server/consumer.go.
    /// </summary>
    public ConsumerState GetConsumerState()
    {
        _mu.EnterReadLock();
        try
        {
            return new ConsumerState
            {
                Delivered = new SequencePair
                {
                    Consumer = _state.Delivered.Consumer,
                    Stream = _state.Delivered.Stream,
                },
                AckFloor = new SequencePair
                {
                    Consumer = _state.AckFloor.Consumer,
                    Stream = _state.AckFloor.Stream,
                },
                Pending = _state.Pending is { Count: > 0 } ? new Dictionary<ulong, Pending>(_state.Pending) : null,
                Redelivered = _state.Redelivered is { Count: > 0 } ? new Dictionary<ulong, ulong>(_state.Redelivered) : null,
            };
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    // -------------------------------------------------------------------------
    // Leadership
    // -------------------------------------------------------------------------

    /// <summary>
    /// Returns true if this server is the current consumer leader.
    /// Mirrors <c>consumer.isLeader</c> in server/consumer.go.
    /// </summary>
    public bool IsLeader()
    {
        _mu.EnterReadLock();
        try { return _isLeader && !_closed; }
        finally { _mu.ExitReadLock(); }
    }

    /// <summary>
    /// Transitions this consumer into or out of the leader role.
    /// Mirrors <c>consumer.setLeader</c> in server/consumer.go.
    /// </summary>
    public void SetLeader(bool isLeader, ulong term)
    {
        _mu.EnterWriteLock();
        try
        {
            _isLeader = isLeader;
            _leaderTerm = term;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal (NatsStream? Stream, IRaftNode? Node) StreamAndNode()
    {
        _mu.EnterReadLock();
        try
        {
            return (_streamRef, _node as IRaftNode);
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal (int Replicas, Exception? Error) Replica()
    {
        _mu.EnterReadLock();
        try
        {
            if (_closed || _streamRef == null)
                return (0, new InvalidOperationException("bad consumer"));

            return (Math.Max(1, Config.Replicas), null);
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal RaftGroup? RaftGroup()
    {
        _mu.EnterReadLock();
        try
        {
            return _assignment?.Group;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void ClearRaftNode()
    {
        _mu.EnterWriteLock();
        try
        {
            _node = null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal IRaftNode? RaftNode()
    {
        _mu.EnterReadLock();
        try
        {
            return _node as IRaftNode;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal Exception? ProcessReplicatedAck(ulong dseq, ulong sseq)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_closed)
                return new InvalidOperationException("consumer closed");

            _state.Delivered.Consumer = Math.Max(_state.Delivered.Consumer, dseq);
            _state.AckFloor.Consumer = Math.Max(_state.AckFloor.Consumer, dseq);
            _state.AckFloor.Stream = Math.Max(_state.AckFloor.Stream, sseq);
            Interlocked.Exchange(ref AckFloor, (long)_state.AckFloor.Stream);
            Exception? noError = null;
            return noError;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool ShouldSendLostQuorum()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_node is not IRaftNode raft || !raft.LostQuorum())
                return false;

            if (DateTime.UtcNow - _lostQuorumSent < TimeSpan.FromSeconds(30))
                return false;

            _lostQuorumSent = DateTime.UtcNow;
            return true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void SetDeliveredStreamSequence(ulong sseq)
    {
        _mu.EnterWriteLock();
        try
        {
            _state.Delivered.Stream = Math.Max(_state.Delivered.Stream, sseq);
            Interlocked.Exchange(ref Delivered, (long)_state.Delivered.Stream);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool IsClosed()
    {
        _mu.EnterReadLock();
        try
        {
            return _closed;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void StepDownRaftNode()
    {
        _mu.EnterReadLock();
        try
        {
            if (_node is IRaftNode raft && raft.Leader())
                raft.StepDown();
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    // -------------------------------------------------------------------------
    // IDisposable
    // -------------------------------------------------------------------------

    public void Dispose()
    {
        _quitCts?.Cancel();
        _quitCts?.Dispose();
        _quitCts = null;
        _mu.Dispose();
    }
}
