namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsStream
{
    internal void SetupMirrorConsumer()
    {
        _mu.EnterWriteLock();
        try
        {
            _mirrorInfo ??= new StreamSourceInfo
            {
                Name = Config.Mirror?.Name ?? Name,
            };
            _mirrorInfo.Active = DateTime.UtcNow;
            _mirrorInfo.Error = null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal StreamSource? StreamSource(string indexName)
    {
        _mu.EnterReadLock();
        try
        {
            return (Config.Sources ?? []).FirstOrDefault(source => string.Equals(source.IndexName, indexName, StringComparison.Ordinal));
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void RetrySourceConsumerAtSeq(string indexName, ulong sequence)
    {
        _mu.EnterWriteLock();
        try
        {
            _sourceStartingSequences[indexName] = sequence;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void CancelSourceConsumer(string indexName)
        => CancelSourceInfo(indexName);

    internal void CancelSourceInfo(string indexName)
    {
        _mu.EnterWriteLock();
        try
        {
            _sources.Remove(indexName);
            _sourceStartingSequences.Remove(indexName);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void SetupSourceConsumer(string indexName, ulong sequence, DateTime requestedAtUtc)
    {
        _mu.EnterWriteLock();
        try
        {
            if (!_sources.TryGetValue(indexName, out var info))
            {
                info = new StreamSourceInfo { Name = indexName };
                _sources[indexName] = info;
            }

            info.Active = requestedAtUtc == default ? DateTime.UtcNow : requestedAtUtc;
            info.Lag = sequence > 0 ? sequence - 1 : 0;
            info.Error = null;
            _sourceStartingSequences[indexName] = sequence;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool TrySetupSourceConsumer(string indexName, ulong sequence, DateTime requestedAtUtc)
    {
        SetupSourceConsumer(indexName, sequence, requestedAtUtc);
        return true;
    }

    internal bool ProcessAllSourceMsgs(string indexName, IReadOnlyList<InMsg> messages)
    {
        var handled = true;
        foreach (var message in messages)
            handled &= ProcessInboundSourceMsg(indexName, message);

        return handled;
    }

    internal void SendFlowControlReply(string replySubject)
    {
        _ = replySubject;
    }

    internal void HandleFlowControl(InMsg message)
    {
        if (!string.IsNullOrWhiteSpace(message.Reply))
            SendFlowControlReply(message.Reply);
    }

    internal bool ProcessInboundSourceMsg(string indexName, InMsg message)
    {
        if (message.IsControlMsg())
        {
            HandleFlowControl(message);
            return true;
        }

        _mu.EnterWriteLock();
        try
        {
            if (!_sources.TryGetValue(indexName, out var info))
                return false;

            info.Active = DateTime.UtcNow;
            info.Lag = info.Lag > 0 ? info.Lag - 1 : 0;
            return true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal static (string Stream, ulong Sequence) StreamAndSeqFromAckReply(string reply)
        => StreamAndSeq(reply);

    internal static (string Stream, ulong Sequence) StreamAndSeq(string reply)
    {
        if (string.IsNullOrWhiteSpace(reply))
            return (string.Empty, 0);

        var tokens = reply.Split('.', StringSplitOptions.RemoveEmptyEntries);
        if (tokens.Length < 2)
            return (string.Empty, 0);

        var stream = tokens[0];
        _ = ulong.TryParse(tokens[^1], out var sequence);
        return (stream, sequence);
    }

    internal void SetStartingSequenceForSources(IDictionary<string, ulong> startingSequences)
    {
        _mu.EnterWriteLock();
        try
        {
            foreach (var pair in startingSequences)
                _sourceStartingSequences[pair.Key] = pair.Value;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void ResetSourceInfo(string indexName)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_sources.TryGetValue(indexName, out var info))
            {
                info.Active = null;
                info.Error = null;
                info.Lag = 0;
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal ulong StartingSequenceForSources(string indexName)
    {
        _mu.EnterReadLock();
        try
        {
            return _sourceStartingSequences.TryGetValue(indexName, out var sequence) ? sequence : 0;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void SetupSourceConsumers()
    {
        _mu.EnterWriteLock();
        try
        {
            foreach (var source in Config.Sources ?? [])
            {
                source.SetIndexName();
                if (!_sources.ContainsKey(source.IndexName))
                {
                    _sources[source.IndexName] = new StreamSourceInfo
                    {
                        Name = source.Name,
                        FilterSubject = source.FilterSubject,
                        External = source,
                    };
                }

                var sequence = source.OptStartSeq > 0 ? source.OptStartSeq : 1;
                _sourceStartingSequences[source.IndexName] = sequence;
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void RetryDisconnectedSyncConsumers()
    {
        _mu.EnterWriteLock();
        try
        {
            if (_mirrorInfo != null && _mirrorInfo.Active == null)
                ScheduleSetupMirrorConsumerRetry();

            foreach (var key in _sources.Keys.ToArray())
            {
                if (_sources[key].Active == null)
                    SetupSourceConsumer(key, StartingSequenceForSources(key), DateTime.UtcNow);
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void ProcessMirrorMsgs(StreamSourceInfo mirrorInfo, IReadOnlyList<InMsg> messages)
    {
        foreach (var message in messages)
            ProcessInboundMirrorMsg(message);

        _mu.EnterWriteLock();
        try
        {
            _mirrorInfo = SourceInfo(mirrorInfo);
            if (_mirrorInfo != null)
                _mirrorInfo.Active = DateTime.UtcNow;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool ProcessInboundMirrorMsg(InMsg message)
    {
        if (message.IsControlMsg())
        {
            HandleFlowControl(message);
            return true;
        }

        _mu.EnterWriteLock();
        try
        {
            if (_mirrorInfo == null)
                return false;

            _mirrorInfo.Active = DateTime.UtcNow;
            _mirrorInfo.Lag = _mirrorInfo.Lag > 0 ? _mirrorInfo.Lag - 1 : 0;
            return true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }
}
