// Copyright 2012-2026 The NATS Authors
// Licensed under the Apache License, Version 2.0

using System.Threading;

namespace ZB.MOM.NatsNet.Server;

internal sealed partial class Raft
{
    public void SendSnapshotToFollower(string peer, byte[] snapshot, ulong term = 0, ulong index = 0)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(peer);
        ArgumentNullException.ThrowIfNull(snapshot);

        _lock.EnterWriteLock();
        try
        {
            var catchup = CatchupFollower(peer, term == 0 ? Term_ : term, index == 0 ? Commit : index);
            catchup.Signal = false;
            InstallSnapshot(snapshot, force: true);
            Commit = Math.Max(Commit, catchup.CIndex);
            Applied_ = Math.Max(Applied_, catchup.CIndex);
            TrackPeer(peer, catchup.CIndex);
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public CatchupState CatchupFollower(string peer, ulong term, ulong index)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(peer);

        _lock.EnterWriteLock();
        try
        {
            var catchup = CreateCatchup();
            catchup.CTerm = term;
            catchup.CIndex = index;
            catchup.PTerm = PTerm;
            catchup.PIndex = PIndex;
            catchup.Sub = peer;
            catchup.Active = DateTime.UtcNow;
            catchup.Signal = true;
            return catchup;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public Entry? LoadEntry(ulong index)
    {
        _lock.EnterReadLock();
        try
        {
            if (index == 0 || !Pae.TryGetValue(index, out var appendEntry) || appendEntry.Entries.Count == 0)
            {
                return null;
            }

            return appendEntry.Entries[0];
        }
        finally
        {
            _lock.ExitReadLock();
        }
    }

    public CommittedEntry? ApplyCommit(ulong commitIndex)
    {
        _lock.EnterWriteLock();
        try
        {
            if (commitIndex <= Applied_)
            {
                return null;
            }

            var committed = NewCommittedEntry(commitIndex);
            for (var index = Applied_ + 1; index <= commitIndex; index++)
            {
                if (Pae.TryGetValue(index, out var appendEntry))
                {
                    committed.Entries.AddRange(appendEntry.Entries);
                }
            }

            Applied_ = commitIndex;
            Commit = Math.Max(Commit, commitIndex);
            ApplyQ_ ??= new ZB.MOM.NatsNet.Server.Internal.IpQueue<CommittedEntry>($"{GroupName}-apply");
            ApplyQ_.Push(committed);
            return committed;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public bool TryCommit(ulong commitIndex) => ApplyCommit(commitIndex) is not null;

    public void TrackResponse(AppendEntryResponse response)
    {
        ArgumentNullException.ThrowIfNull(response);

        _lock.EnterWriteLock();
        try
        {
            if (State() != RaftState.Leader)
            {
                return;
            }

            TrackPeer(response.Peer, response.Index);
            if (response.Success)
            {
                if (!Acks.TryGetValue(response.Index, out var acks))
                {
                    acks = new Dictionary<string, bool>(StringComparer.Ordinal);
                    Acks[response.Index] = acks;
                }

                acks[response.Peer] = true;
                TryCommit(response.Index);
            }
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void AdjustClusterSizeAndQuorum()
    {
        _lock.EnterWriteLock();
        try
        {
            Csz = Math.Max(Peers_.Count + 1, 1);
            Qn = (Csz / 2) + 1;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void TrackPeer(string peer, ulong index)
    {
        ArgumentException.ThrowIfNullOrWhiteSpace(peer);

        _lock.EnterWriteLock();
        try
        {
            if (!Peers_.TryGetValue(peer, out var state))
            {
                state = new Lps();
                Peers_[peer] = state;
            }

            state.Li = Math.Max(state.Li, index);
            state.Ts = DateTime.UtcNow;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void RunAsCandidate()
    {
        SwitchToCandidate();
        RequestVote();
    }

    public void HandleAppendEntry(AppendEntry appendEntry)
    {
        ProcessAppendEntries(appendEntry);
    }

    public void CancelCatchup()
    {
        _lock.EnterWriteLock();
        try
        {
            Catchup = null;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public bool CatchupStalled(TimeSpan? threshold = null)
    {
        _lock.EnterReadLock();
        try
        {
            if (Catchup is null)
            {
                return false;
            }

            var timeout = threshold ?? TimeSpan.FromSeconds(10);
            return DateTime.UtcNow - Catchup.Active > timeout;
        }
        finally
        {
            _lock.ExitReadLock();
        }
    }

    public CatchupState CreateCatchup()
    {
        _lock.EnterWriteLock();
        try
        {
            Catchup ??= new CatchupState();
            Catchup.Active = DateTime.UtcNow;
            return Catchup;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public bool SendCatchupSignal()
    {
        _lock.EnterWriteLock();
        try
        {
            var catchup = CreateCatchup();
            catchup.Signal = true;
            catchup.Active = DateTime.UtcNow;
            return true;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void CancelCatchupSignal()
    {
        _lock.EnterWriteLock();
        try
        {
            if (Catchup is not null)
            {
                Catchup.Signal = false;
            }
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void TruncateWAL(ulong index)
    {
        _lock.EnterWriteLock();
        try
        {
            foreach (var key in Pae.Keys.Where(key => key > index).ToArray())
            {
                Pae.Remove(key);
            }

            PIndex = Math.Min(PIndex, index);
            Commit = Math.Min(Commit, index);
            Applied_ = Math.Min(Applied_, index);
            Processed_ = Math.Min(Processed_, index);
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void ResetWAL()
    {
        _lock.EnterWriteLock();
        try
        {
            Pae.Clear();
            Acks.Clear();
            PIndex = 0;
            PTerm = 0;
            Commit = 0;
            Applied_ = 0;
            Processed_ = 0;
            WalBytes = 0;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }

    public void UpdateLeader(string leaderId)
    {
        _lock.EnterWriteLock();
        try
        {
            LeaderId = leaderId ?? string.Empty;
            if (!string.IsNullOrWhiteSpace(LeaderId) &&
                !string.Equals(LeaderId, Id, StringComparison.Ordinal) &&
                State() == RaftState.Leader)
            {
                StateValue = (int)RaftState.Follower;
            }

            Interlocked.Exchange(ref HasLeaderV, string.IsNullOrWhiteSpace(LeaderId) ? 0 : 1);
            if (!string.IsNullOrWhiteSpace(LeaderId))
            {
                Interlocked.Exchange(ref PLeaderV, 1);
            }

            Active = DateTime.UtcNow;
        }
        finally
        {
            _lock.ExitWriteLock();
        }
    }
}
