using System.Text;
using ZB.MOM.NatsNet.Server.Internal;

namespace ZB.MOM.NatsNet.Server;

internal sealed partial class NatsStream
{
    internal void SubscribeToStream()
    {
        foreach (var subject in Config.Subjects ?? [])
            _ = SubscribeInternal(subject, handler: null);
    }

    internal void SubscribeToDirect()
    {
        _mu.EnterWriteLock();
        try
        {
            _allowDirectSubscription = true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void UnsubscribeToDirect()
    {
        _mu.EnterWriteLock();
        try
        {
            _allowDirectSubscription = false;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void SubscribeToMirrorDirect()
    {
        _mu.EnterWriteLock();
        try
        {
            _allowMirrorDirectSubscription = true;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void UnsubscribeToMirrorDirect()
    {
        _mu.EnterWriteLock();
        try
        {
            _allowMirrorDirectSubscription = false;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void StopSourceConsumers()
    {
        _mu.EnterWriteLock();
        try
        {
            _sources.Clear();
            _sourceStartingSequences.Clear();
            _mirrorInfo = null;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void RemoveInternalConsumer(string subject)
    {
        _ = UnsubscribeInternal(subject);
    }

    internal void UnsubscribeToStream()
    {
        foreach (var subject in Config.Subjects ?? [])
            _ = UnsubscribeInternal(subject);
    }

    internal void DeleteInflightBatches(bool preserveState)
    {
        _mu.EnterWriteLock();
        try
        {
            _inflightBatches.Clear();
            if (!preserveState)
                DeleteBatchApplyState();
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal void DeleteBatchApplyState()
    {
        _mu.EnterWriteLock();
        try
        {
            _inflightBatches.Clear();
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal (Subscription? Subscription, Exception? Error) SubscribeInternal(string subject, Action<InMsg>? handler)
    {
        _ = handler;

        if (string.IsNullOrWhiteSpace(subject))
            return (null, new ArgumentException("subject required", nameof(subject)));

        _mu.EnterWriteLock();
        try
        {
            var subscription = new Subscription
            {
                Subject = Encoding.ASCII.GetBytes(subject),
                Sid = Encoding.ASCII.GetBytes(subject),
            };

            _internalSubscriptions[subject] = subscription;
            return (subscription, null);
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }

    internal (Subscription? Subscription, Exception? Error) QueueSubscribeInternal(string subject, string queue, Action<InMsg>? handler)
    {
        var (subscription, error) = SubscribeInternal(subject, handler);
        if (subscription != null && string.IsNullOrWhiteSpace(queue) is false)
            subscription.Queue = Encoding.ASCII.GetBytes(queue);

        return (subscription, error);
    }

    internal Exception? UnsubscribeInternal(string subject)
    {
        if (string.IsNullOrWhiteSpace(subject))
            return new ArgumentException("subject required", nameof(subject));

        _mu.EnterWriteLock();
        try
        {
            if (_internalSubscriptions.TryGetValue(subject, out var subscription))
            {
                subscription.Close();
                _internalSubscriptions.Remove(subject);
            }

            Exception? error = null;
            return error;
        }
        finally
        {
            _mu.ExitWriteLock();
        }
    }
}
