namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsStream
{
    private static readonly TimeSpan RetryBackoff = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan RetryMaximum = TimeSpan.FromMinutes(2);
    private static readonly TimeSpan SourceConsumerRetryThreshold = TimeSpan.FromSeconds(10);

    internal Exception? UpdateWithAdvisory(StreamConfig config, bool sendAdvisory, bool pedantic)
    {
        _ = pedantic;

        if (config == null)
            return new ArgumentNullException(nameof(config));

        UpdateConfig(config);
        Exception? error = null;
        if (sendAdvisory)
            SendUpdateAdvisoryLocked();

        return error;
    }

    internal string GetCfgName()
    {
        _mu.EnterReadLock();
        try
        {
            return Config.Name ?? string.Empty;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal (ulong Purged, Exception? Error) PurgeLocked(StreamPurgeRequest? request, bool needLock)
    {
        if (needLock)
            _mu.EnterWriteLock();

        try
        {
            if (_closed)
                return (0UL, new InvalidOperationException("stream closed"));

            if (_sealed)
                return (0UL, new InvalidOperationException("sealed stream"));

            if (Store == null)
                return (0UL, new InvalidOperationException("stream store unavailable"));

            var result = request == null
                ? Store.Purge()
                : Store.PurgeEx(request.Filter ?? string.Empty, request.Sequence, request.Keep);
            SyncCountersFromState(Store.State());
            return result;
        }
        finally
        {
            if (needLock)
                _mu.ExitWriteLock();
        }
    }

    internal (bool Removed, Exception? Error) RemoveMsg(ulong sequence)
        => DeleteMsg(sequence);

    internal (bool Removed, Exception? Error) DeleteMsg(ulong sequence)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_closed)
                return (false, new InvalidOperationException("stream closed"));

            if (Store == null)
                return (false, new InvalidOperationException("stream store unavailable"));

            var result = Store.RemoveMsg(sequence);
            if (result.Error == null)
                SyncCountersFromState(Store.State());
            return result;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal (bool Removed, Exception? Error) EraseMsg(ulong sequence)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_closed)
                return (false, new InvalidOperationException("stream closed"));

            if (Store == null)
                return (false, new InvalidOperationException("stream store unavailable"));

            var result = Store.EraseMsg(sequence);
            if (result.Error == null)
                SyncCountersFromState(Store.State());
            return result;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal bool IsMirror()
    {
        _mu.EnterReadLock();
        try
        {
            return _isMirror || Config.Mirror != null;
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal StreamSourceInfo[] SourcesInfo()
    {
        _mu.EnterReadLock();
        try
        {
            return _sources.Values.Select(SourceInfo).Where(static info => info != null).Cast<StreamSourceInfo>().ToArray();
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal StreamSourceInfo? SourceInfo(StreamSourceInfo? sourceInfo)
    {
        if (sourceInfo == null)
            return null;

        return new StreamSourceInfo
        {
            Name = sourceInfo.Name,
            Lag = sourceInfo.Lag,
            FilterSubject = sourceInfo.FilterSubject,
            Active = sourceInfo.Active,
            Error = sourceInfo.Error,
            External = sourceInfo.External == null
                ? null
                : new StreamSource
                {
                    Name = sourceInfo.External.Name,
                    FilterSubject = sourceInfo.External.FilterSubject,
                    SubjectTransforms = sourceInfo.External.SubjectTransforms,
                    External = sourceInfo.External.External,
                },
        };
    }

    internal StreamSourceInfo? MirrorInfo()
    {
        _mu.EnterReadLock();
        try
        {
            return SourceInfo(_mirrorInfo);
        }
        finally
        {
            _mu.ExitReadLock();
        }
    }

    internal void SetMirrorErr(JsApiError? error)
    {
        _mu.EnterWriteLock();
        try
        {
            if (_mirrorInfo != null)
                _mirrorInfo.Error = error?.Description;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void CancelMirrorConsumer()
    {
        _mu.EnterWriteLock();
        try
        {
            _mirrorConsumerSetupTimer?.Dispose();
            _mirrorConsumerSetupTimer = null;
            if (_mirrorInfo != null)
            {
                _mirrorInfo.Active = null;
                _mirrorInfo.Error = null;
            }
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal Exception? RetryMirrorConsumer()
    {
        CancelMirrorConsumer();
        Exception? error = null;
        return error;
    }

    internal void SkipMsgs(ulong start, ulong end)
    {
        _mu.EnterWriteLock();
        try
        {
            if (Store == null || start > end)
                return;

            var count = (end - start) + 1;
            Store.SkipMsgs(start, count);
            SetLastSeq(end);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal static TimeSpan CalculateRetryBackoff(int failures)
    {
        var backoff = TimeSpan.FromTicks(RetryBackoff.Ticks * Math.Max(1, failures * 2));
        return backoff > RetryMaximum ? RetryMaximum : backoff;
    }

    internal void ScheduleSetupMirrorConsumerRetry()
    {
        _mu.EnterWriteLock();
        try
        {
            var lastAttempt = _mirrorInfo?.Active ?? DateTime.UtcNow - SourceConsumerRetryThreshold;
            var next = SourceConsumerRetryThreshold - (DateTime.UtcNow - lastAttempt);
            if (next < TimeSpan.Zero)
                next = TimeSpan.Zero;

            var failures = _mirrorInfo == null ? 1 : (int)Math.Min(_mirrorInfo.Lag, int.MaxValue);
            next += CalculateRetryBackoff(failures);
            next += TimeSpan.FromMilliseconds(Random.Shared.Next(100, 201));

            _mirrorConsumerSetupTimer?.Dispose();
            _mirrorConsumerSetupTimer = new Timer(
                static state =>
                {
                    if (state is NatsStream stream)
                        stream.RetryMirrorConsumer();
                },
                this,
                next,
                Timeout.InfiniteTimeSpan);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }
}
