using System.Text.Json;
using IronSnappy;

namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsStream
{
    internal bool SupportsBinarySnapshot()
    {
        _mu.EnterReadLock();
        try
        {
            return SupportsBinarySnapshotLocked();
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal bool SupportsBinarySnapshotLocked()
    {
        var raftNode = _node as IRaftNode;
        var server = Account?.Server as NatsServer;
        if (raftNode == null || server == null)
            return false;

        var ourId = raftNode.ID();
        foreach (var peer in raftNode.Peers())
        {
            if (string.Equals(peer.Id, ourId, StringComparison.Ordinal))
                continue;

            if (server.GetNodeInfo(peer.Id) is { BinarySnapshots: false })
                return false;
        }

        return true;
    }

    internal byte[]? StateSnapshot()
    {
        _mu.EnterReadLock();
        try
        {
            return StateSnapshotLocked();
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal byte[]? StateSnapshotLocked()
    {
        if (Store == null)
            return null;

        if (SupportsBinarySnapshotLocked())
        {
            var (encoded, error) = Store.EncodedStreamState(GetCLFS());
            return error == null ? encoded : null;
        }

        var state = Store.State();
        var snapshot = new StreamSnapshot
        {
            Msgs = state.Msgs,
            Bytes = state.Bytes,
            FirstSeq = state.FirstSeq,
            LastSeq = state.LastSeq,
            Failed = GetCLFS(),
            Deleted = state.Deleted,
        };
        return JsonSerializer.SerializeToUtf8Bytes(snapshot);
    }

    internal Exception? ProcessClusteredInboundMsg(string subject, string reply, byte[]? header, byte[]? message, object? msgTrace, bool sourced)
    {
        _ = msgTrace;
        if (string.IsNullOrWhiteSpace(subject))
            return new ArgumentException("subject is required", nameof(subject));

        _mu.EnterUpgradeableReadLock();
        try
        {
            var raftNode = _node as IRaftNode;
            var canRespond = !Config.NoAck && !string.IsNullOrWhiteSpace(reply);
            if (raftNode == null)
            {
                return ProcessJetStreamMsg(subject, reply, header, message, 0, 0, msgTrace, sourced, canRespond);
            }

            if (!IsLeader())
            {
                return new InvalidOperationException(JsApiErrors.NewJSClusterNotLeaderError().Description ?? "stream is not cluster leader");
            }

            if (Config.Sealed)
            {
                if (canRespond)
                {
                    var response = new JSPubAckResponse
                    {
                        Stream = Name,
                        PubAckError = JsApiErrors.NewJSStreamSealedError(),
                    };
                    _outq.SendMsg(reply, JsonSerializer.SerializeToUtf8Bytes(response));
                }

                return new InvalidOperationException(JsApiErrors.NewJSStreamSealedError().Description ?? "stream is sealed");
            }

            _mu.EnterWriteLock();
            try
            {
                if (_clseq == 0)
                    _clseq = (ulong)Math.Max(0, Interlocked.Read(ref LastSeq)) + _clfs;
                _clseq++;

                var encoded = JetStreamCluster.EncodeStreamMsgAllowCompress(
                    subject,
                    reply,
                    header,
                    message,
                    _clseq,
                    DateTimeOffset.UtcNow.ToUnixTimeMilliseconds() * 1_000_000L,
                    sourced);

                raftNode.Propose(encoded);
                TrackReplicationTraffic(raftNode, encoded.Length, Math.Max(1, Config.Replicas));
            }
            finally
            {
                _mu.ExitWriteLock();
            }

            return null;
        }
        finally
        {
            _mu.ExitUpgradeableReadLock();
        }
    }

    internal object? GetAndDeleteMsgTrace(ulong sequence)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!_msgTraceBySeq.TryGetValue(sequence, out var trace))
                return null;

            _msgTraceBySeq.Remove(sequence);
            return trace;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal StreamSyncRequest? CalculateSyncRequest(StreamState? state, StreamReplicatedState? snapshot, ulong index)
    {
        if (state == null || snapshot == null || _node is not IRaftNode raftNode)
            return null;
        if (state.LastSeq >= snapshot.LastSeq)
            return null;

        return new StreamSyncRequest
        {
            FirstSeq = state.LastSeq + 1,
            LastSeq = snapshot.LastSeq,
            Peer = raftNode.ID(),
            DeleteRangesOk = true,
            MinApplied = index,
        };
    }

    internal void ProcessSnapshotDeletes(StreamReplicatedState snapshot)
    {
        if (Store == null)
            return;

        _mu.EnterWriteLock();
        try
        {
            var state = new StreamState();
            Store.FastState(state);
            if (snapshot.FirstSeq > state.FirstSeq)
            {
                Store.Compact(snapshot.FirstSeq);
                Store.FastState(state);
                Interlocked.Exchange(ref LastSeq, (long)state.LastSeq);
                ClearAllPreAcksBelowFloor(state.FirstSeq);
            }

            if (snapshot.Deleted.Count > 0)
                Store.SyncDeleted(snapshot.Deleted);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void SetCatchupPeer(string peer, ulong lag)
    {
        if (string.IsNullOrWhiteSpace(peer))
            return;

        _mu.EnterWriteLock();
        try
        {
            _catchupPeers ??= new Dictionary<string, ulong>(StringComparer.Ordinal);
            _catchupPeers[peer] = lag;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void UpdateCatchupPeer(string peer) => DecrementCatchupPeer(peer, 1);

    internal void DecrementCatchupPeer(string peer, ulong decrementBy)
    {
        if (string.IsNullOrWhiteSpace(peer) || decrementBy == 0)
            return;

        _mu.EnterWriteLock();
        try
        {
            if (_catchupPeers == null || !_catchupPeers.TryGetValue(peer, out var lag) || lag == 0)
                return;
            _catchupPeers[peer] = lag > decrementBy ? lag - decrementBy : 0;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void ClearCatchupPeer(string peer)
    {
        _mu.EnterWriteLock();
        try
        {
            _catchupPeers?.Remove(peer);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void ClearAllCatchupPeers()
    {
        _mu.EnterWriteLock();
        try
        {
            _catchupPeers = null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal ulong LagForCatchupPeer(string peer)
    {
        _mu.EnterReadLock();
        try
        {
            if (_catchupPeers == null || !_catchupPeers.TryGetValue(peer, out var lag))
                return 0;
            return lag;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal bool HasCatchupPeers()
    {
        _mu.EnterReadLock();
        try
        {
            return _catchupPeers is { Count: > 0 };
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void SetCatchingUp() => Interlocked.Exchange(ref _catchingUp, 1);

    internal void ClearCatchingUp() => Interlocked.Exchange(ref _catchingUp, 0);

    internal bool IsCatchingUp() => Interlocked.CompareExchange(ref _catchingUp, 0, 0) == 1;

    internal bool IsCurrent()
    {
        if (_node is not IRaftNode raftNode)
            return true;
        return raftNode.Current() && !IsCatchingUp();
    }

    internal Exception? ProcessSnapshot(StreamReplicatedState snapshot, ulong index)
    {
        ProcessSnapshotDeletes(snapshot);
        SetCLFS(snapshot.Failed);

        if (Store == null || _assignment == null || _node is not IRaftNode raftNode)
            return new InvalidOperationException("stream has been stopped");

        var state = new StreamState();
        Store.FastState(state);
        var syncRequest = CalculateSyncRequest(state, snapshot, index);
        if (syncRequest == null)
            return null;

        try
        {
            raftNode.PauseApply();
            SetCatchingUp();
            RunCatchup(string.Empty, syncRequest);
            return null;
        }
        catch (Exception ex)
        {
            return ex;
        }
        finally
        {
            ClearCatchingUp();
            raftNode.ResumeApply();
        }
    }

    internal (ulong Sequence, Exception? Error) ProcessCatchupMsg(byte[] encodedMessage)
    {
        if (encodedMessage == null || encodedMessage.Length == 0)
            return (0, new InvalidOperationException("bad catchup msg"));
        if (Store == null)
            return (0, new InvalidOperationException("store not initialized"));

        var operation = (EntryOp)encodedMessage[0];
        var payload = encodedMessage.AsSpan(1);

        if (operation == EntryOp.DeleteRangeOp)
        {
            var (deleteRange, decodeError) = JetStreamCluster.DecodeDeleteRange(payload);
            if (decodeError != null || deleteRange == null)
                return (0, new InvalidOperationException("bad catchup msg"));

            _mu.EnterWriteLock();
            try
            {
                if (_preAcks.Count > 0)
                {
                    for (ulong seq = deleteRange.First; seq < deleteRange.First + deleteRange.Num; seq++)
                        ClearAllPreAcks(seq);
                }

                Store.SkipMsgs(deleteRange.First, deleteRange.Num);
                var last = deleteRange.First + deleteRange.Num - 1;
                SetLastSeq(last);
                return (last, null);
            }
            catch (Exception ex)
            {
                return (0, ex);
            }
            finally
            {
                _mu.ExitWriteLock();
            }
        }

        if (operation == EntryOp.CompressedStreamMsgOp)
            payload = Snappy.Decode(payload);

        var (subject, _, header, message, sequence, timestamp, _, decodeStreamError) = JetStreamCluster.DecodeStreamMsg(payload);
        if (decodeStreamError != null)
            return (0, new InvalidOperationException("bad catchup msg"));

        if (!string.IsNullOrEmpty(subject) || timestamp != 0)
            Store.StoreRawMsg(subject, header, message, sequence, timestamp, ttl: 0, discardNewCheck: false);
        else
            Store.SkipMsg(sequence);

        SetLastSeq(sequence);
        return (sequence, null);
    }

    internal void FlushAllPending() => Store?.FlushAllPending();

    internal void HandleClusterSyncRequest(object? sub, ClientConnection? client, Account? account, string subject, string reply, byte[] message)
    {
        _ = sub;
        _ = client;
        _ = account;
        _ = subject;

        StreamSyncRequest? request;
        try
        {
            request = JsonSerializer.Deserialize<StreamSyncRequest>(message);
        }
        catch
        {
            return;
        }

        if (request == null)
            return;

        _ = Task.Run(() => RunCatchup(reply, request));
    }

    internal void RunCatchup(string sendSubject, StreamSyncRequest request)
    {
        if (Store == null)
            return;

        if (request.LastSeq < request.FirstSeq)
            return;

        SetCatchupPeer(request.Peer, request.LastSeq - request.FirstSeq);
        try
        {
            var state = new StreamState();
            Store.FastState(state);
            if (state.LastSeq < request.FirstSeq)
                return;

            // Current C# port keeps catchup streaming minimal: this method updates
            // catchup peer accounting and relies on existing replication apply paths.
            ClearCatchupPeer(request.Peer);
        }
        finally
        {
            if (!string.IsNullOrWhiteSpace(sendSubject))
                ClearCatchupPeer(request.Peer);
        }
    }

    internal void CheckClusterInfo(ClusterInfo? clusterInfo)
    {
        if (clusterInfo?.Replicas == null || clusterInfo.Replicas.Length == 0)
            return;

        foreach (var replica in clusterInfo.Replicas)
        {
            var peer = NatsServer.GetHash(replica.Name);
            var lag = LagForCatchupPeer(peer);
            if (lag == 0)
                continue;

            replica.Current = false;
            replica.Lag = lag;
        }
    }

    internal void HandleClusterStreamInfoRequest(object? sub, ClientConnection? client, Account? account, string subject, string reply, byte[] message)
    {
        _ = sub;
        _ = client;
        _ = account;
        _ = subject;
        _ = message;
        _ = Task.Run(() => ProcessClusterStreamInfoRequest(reply));
    }

    internal void ProcessClusterStreamInfoRequest(string reply)
    {
        _mu.EnterReadLock();
        try
        {
            if (string.IsNullOrWhiteSpace(reply))
                return;

            var streamInfo = new StreamInfo
            {
                Created = CreatedTime(),
                State = State(),
                Config = Config.Clone(),
                Cluster = null,
                Sources = SourcesInfo(),
                Mirror = MirrorInfo(),
            };

            if (HasCatchupPeers())
                CheckClusterInfo(streamInfo.Cluster);

            _outq.SendMsg(reply, JsonSerializer.SerializeToUtf8Bytes(streamInfo));
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }
}
