﻿/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

using System;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;
using Apache.NMS.AMQP.Message;
using Apache.NMS.AMQP.Meta;
using Apache.NMS.AMQP.Util;
using Apache.NMS.AMQP.Util.Synchronization;
using Apache.NMS.Util;

namespace Apache.NMS.AMQP.Provider.Failover
{
    public class FailoverProvider : IProvider, IProviderListener
    {
        private readonly object SyncRoot = new object();

        private static int UNDEFINED = -1;

        public static int DEFAULT_INITIAL_RECONNECT_DELAY = 0;
        public static long DEFAULT_RECONNECT_DELAY = 10;
        public static double DEFAULT_RECONNECT_BACKOFF_MULTIPLIER = 2.0d;
        public static long DEFAULT_MAX_RECONNECT_DELAY = (long) Math.Round(TimeSpan.FromSeconds(30).TotalMilliseconds);
        public static int DEFAULT_STARTUP_MAX_RECONNECT_ATTEMPTS = UNDEFINED;
        public static int DEFAULT_MAX_RECONNECT_ATTEMPTS = UNDEFINED;
        public static bool DEFAULT_USE_RECONNECT_BACKOFF = true;
        public static int DEFAULT_WARN_AFTER_RECONNECT_ATTEMPTS = 10;
        public static bool DEFAULT_RANDOMIZE_ENABLED = false;

        private readonly ReconnectControls reconnectControl;
        private readonly FailoverUriPool uris;

        private readonly AtomicBool closed = new AtomicBool();
        private readonly Atomic<bool> failed = new Atomic<bool>();

        private long requestTimeout;
        private long sendTimeout;

        private Uri connectedUri;
        private NmsConnectionInfo connectionInfo;
        private IProvider provider;
        private IProviderListener listener;

        private List<FailoverRequest> requests = new List<FailoverRequest>();

        internal IProvider ActiveProvider => provider;

        public FailoverProvider(IEnumerable<Uri> uris)
        {
            this.uris = new FailoverUriPool(uris);
            reconnectControl = new ReconnectControls(this);
        }

        public long InitialReconnectDelay { get; set; } = DEFAULT_INITIAL_RECONNECT_DELAY;
        public long ReconnectDelay { get; set; } = DEFAULT_RECONNECT_DELAY;
        public bool UseReconnectBackOff { get; set; } = DEFAULT_USE_RECONNECT_BACKOFF;
        public double ReconnectBackOffMultiplier { get; set; } = DEFAULT_RECONNECT_BACKOFF_MULTIPLIER;
        public long MaxReconnectDelay { get; set; } = DEFAULT_MAX_RECONNECT_DELAY;
        public int StartupMaxReconnectAttempts { get; set; } = DEFAULT_STARTUP_MAX_RECONNECT_ATTEMPTS;
        public int MaxReconnectAttempts { get; set; } = DEFAULT_MAX_RECONNECT_ATTEMPTS;
        public int WarnAfterReconnectAttempts { get; set; } = DEFAULT_WARN_AFTER_RECONNECT_ATTEMPTS;
        public Uri RemoteUri => connectedUri;

        public void Start()
        {
            CheckClosed();

            if (listener == null)
            {
                throw new IllegalStateException("No ProviderListener registered.");
            }
        }

        public Task Connect(NmsConnectionInfo info)
        {
            CheckClosed();

            requestTimeout = info.RequestTimeout;
            sendTimeout = info.SendTimeout;

            connectionInfo = info;
            Tracer.Debug("Initiating initial connection attempt task");
            return TriggerReconnectionAttempt();
        }

        private Task TriggerReconnectionAttempt()
        {
            if (closed)
                return Task.CompletedTask;

            return reconnectControl.ScheduleReconnect(Reconnect);

            async Task Reconnect()
            {
                IProvider provider = null;
                Exception failure = null;
                long reconnectAttempts = reconnectControl.RecordNextAttempt();

                try
                {
                    if (uris.Any())
                    {
                        for (int i = 0; i < uris.Size(); i++)
                        {
                            var target = uris.GetNext();
                            if (target == null)
                            {
                                Tracer.Debug("Failover URI collection unexpectedly modified during connection attempt.");
                                continue;
                            }

                            try
                            {
                                Tracer.Debug($"Connection attempt:[{reconnectAttempts}] to: {target.Scheme}://{target.Host}:{target.Port} in-progress");
                                provider = ProviderFactory.Create(target);
                                await provider.Connect(connectionInfo).Await();
                                await InitializeNewConnection(provider).Await();
                                return;
                            }
                            catch (Exception e)
                            {
                                Tracer.Info($"Connection attempt:[{reconnectAttempts}] to: {target.Scheme}://{target.Host}:{target.Port} failed");
                                failure = e;
                                try
                                {
                                    provider?.Close();
                                }
                                catch
                                {
                                }
                                finally
                                {
                                    provider = null;
                                }
                            }
                        }
                    }
                    else
                    {
                        Tracer.Debug("No remote URI available to connect to in failover list");
                        // TODO Handle this one.
                        failure = new IOException("No remote URI available for reconnection during connection attempt: " + reconnectAttempts);
                    }
                }
                catch (Exception unknownFailure)
                {
                    Tracer.Warn($"Connection attempt:[{reconnectAttempts}] failed abnormally.");
                    failure = failure ?? unknownFailure;
                }
                finally
                {
                    if (provider == null)
                    {
                        Tracer.Debug($"Connection attempt:[{reconnectControl.ReconnectAttempts}] failed error: {failure?.Message}");
                        if (!reconnectControl.IsReconnectAllowed(failure))
                        {
                            ReportReconnectFailure(failure);
                        }
                        else
                        {
                            await reconnectControl.ScheduleReconnect(Reconnect).Await();
                        }
                    }
                }
            }
        }

        private void ReportReconnectFailure(Exception lastFailure)
        {
            Tracer.Error($"Failed to connect after: {reconnectControl.ReconnectAttempts} attempt(s)");
            if (failed.CompareAndSet(false, true))
            {
                if (lastFailure == null)
                {
                    lastFailure = new IOException($"Failed to connect after: {reconnectControl.ReconnectAttempts} attempt(s)");
                }

                var exception = NMSExceptionSupport.Create(lastFailure);
                listener.OnConnectionFailure(exception);
                throw exception;
            }
        }

        private async Task InitializeNewConnection(IProvider provider)
        {
            if (closed)
            {
                try
                {
                    provider.Close();
                }
                catch (Exception e)
                {
                    Tracer.Debug($"Ignoring failure to close failed provider: {provider} {e.Message}");
                }
                return;
            }

            this.provider = provider;
            this.provider.SetProviderListener(this);
            this.connectedUri = provider.RemoteUri;

            if (reconnectControl.IsRecoveryRequired())
            {
                Tracer.Debug($"Signalling connection recovery: {provider}");

                // Allow listener to recover its resources
                await listener.OnConnectionRecovery(provider).Await();

                // Restart consumers, send pull commands, etc.
                await listener.OnConnectionRecovered(provider).Await();

                // Let the client know that connection has restored.
                listener.OnConnectionRestored(connectedUri);

                // If we try to run pending requests right after the connection is reestablished 
                // it will result in timeout on the first send request
                await Task.Delay(50).Await();

                foreach (FailoverRequest request in GetPendingRequests())
                {
                    await request.Run().Await();
                }

                reconnectControl.ConnectionEstablished();
            }
            else
            {
                listener.OnConnectionEstablished(connectedUri);
                reconnectControl.ConnectionEstablished();
            }
        }

        public override string ToString()
        {
            return "FailoverProvider: " + (connectedUri == null ? "unconnected" : connectedUri.ToString());
        }

        public void Close()
        {
            if (closed.CompareAndSet(false, true))
            {
                try
                {
                    provider?.Close();
                }
                catch (Exception e)
                {
                    Tracer.Warn("Error caught while closing Provider: " + e.Message);
                }
            }
        }

        public async Task CloseAsync()
        {
            if (closed.CompareAndSet(false, true))
            {
                try
                {
                    if (provider != null) await provider.CloseAsync().Await();
                }
                catch (Exception e)
                {
                    Tracer.Warn("Error caught while closing Provider: " + e.Message);
                }
            }
        }

        public void SetProviderListener(IProviderListener providerListener)
        {
            CheckClosed();

            listener = providerListener;
        }

        public Task CreateResource(INmsResource resourceInfo)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => provider.CreateResource(resourceInfo),
                Name = nameof(CreateResource)
            };

            request.Run();

            return request.Task;
        }

        public Task DestroyResource(INmsResource resourceInfo)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => activeProvider.DestroyResource(resourceInfo),
                Name = nameof(DestroyResource),

                // Allow this to succeed, resource won't get recreated on reconnect.
                SucceedsWhenOffline = true
            };

            request.Run();

            return request.Task;
        }

        public Task StartResource(INmsResource resourceInfo)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => activeProvider.StartResource(resourceInfo),
                Name = nameof(StartResource)
            };

            request.Run();

            return request.Task;
        }

        public Task StopResource(INmsResource resourceInfo)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => activeProvider.StopResource(resourceInfo),
                Name = nameof(StopResource)
            };

            request.Run();

            return request.Task;
        }

        public Task Recover(NmsSessionId sessionId)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => activeProvider.Recover(sessionId),
                SucceedsWhenOffline = true,
                Name = nameof(Recover)
            };

            request.Run();

            return request.Task;
        }

        public Task Acknowledge(NmsSessionId sessionId, AckType ackType)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => activeProvider.Acknowledge(sessionId, ackType),
                FailureWhenOffline = true,
                Name = nameof(Acknowledge)
            };

            request.Run();

            return request.Task;
        }

        public Task Acknowledge(InboundMessageDispatch envelope, AckType ackType)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, requestTimeout)
            {
                DoTask = activeProvider => activeProvider.Acknowledge(envelope, ackType),
                FailureWhenOffline = true,
                Name = nameof(Acknowledge)
            };

            request.Run();

            return request.Task;
        }

        public INmsMessageFactory MessageFactory => provider.MessageFactory;

        public long SendTimeout => sendTimeout;

        public Task Send(OutboundMessageDispatch envelope)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, SendTimeout)
            {
                DoTask = activeProvider => activeProvider.Send(envelope),
                Name = nameof(Send)
            };

            request.Run();

            return request.Task;
        }

        public Task Unsubscribe(string name)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, SendTimeout)
            {
                DoTask = activeProvider => activeProvider.Unsubscribe(name),
                Name = nameof(Unsubscribe)
            };

            request.Run();

            return request.Task;
        }

        public Task Rollback(NmsTransactionInfo transactionInfo, NmsTransactionInfo nextTransactionInfo)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, SendTimeout)
            {
                DoTask = activeProvider => activeProvider.Rollback(transactionInfo, nextTransactionInfo),
                Name = nameof(Rollback),
                SucceedsWhenOffline = true
            };

            request.Run();

            return request.Task;
        }

        public Task Commit(NmsTransactionInfo transactionInfo, NmsTransactionInfo nextTransactionInfo)
        {
            CheckClosed();

            FailoverRequest request = new FailoverRequest(this, SendTimeout)
            {
                DoTask = activeProvider => activeProvider.Commit(transactionInfo, nextTransactionInfo),
                Name = nameof(Commit),
                FailureWhenOffline = true
            };

            request.Run();

            return request.Task;
        }

        public void OnInboundMessage(InboundMessageDispatch envelope)
        {
            if (closed)
                return;

            listener.OnInboundMessage(envelope);
        }

        public void OnConnectionFailure(NMSException exception)
        {
            Tracer.Debug($"Failover: the provider reports failure: {exception.Message}");
            HandleProviderError(provider, exception);
        }

        public Task OnConnectionRecovery(IProvider provider)
        {
            return Task.CompletedTask;
        }

        public void OnConnectionEstablished(Uri remoteUri)
        {
        }

        public Task OnConnectionRecovered(IProvider provider)
        {
            return Task.CompletedTask;
        }

        public void OnConnectionRestored(Uri remoteUri)
        {
        }

        public void OnConnectionInterrupted(Uri failedUri)
        {
        }

        public void OnResourceClosed(INmsResource resourceInfo, Exception error)
        {
            if (closed)
                return;

            listener.OnResourceClosed(resourceInfo, error);
        }

        internal void HandleProviderError(IProvider provider, NMSException cause)
        {
            if (closed)
                return;

            lock (SyncRoot)
            {
                // It is possible that another failed request signaled an error for the same provider
                // and we already cleaned up the old failed provider and scheduled a reconnect that
                // has already succeeded, so we need to ensure that we don't kill a valid provider.
                if (provider == this.provider)
                {
                    Tracer.Debug($"handling Provider failure: {cause.ToString()}");
                    this.provider = null;
                    provider.SetProviderListener(null);
                    Uri failedUri = provider.RemoteUri;
                    try
                    {
                        provider.Close();
                    }
                    catch (Exception exception)
                    {
                        Tracer.Debug($"Caught exception while closing failed provider: {exception.Message}");
                    }

                    if (reconnectControl.IsReconnectAllowed(cause))
                    {
                        // Start watching for request timeouts while we are offline, unless we already are.
                        foreach (FailoverRequest failoverRequest in GetPendingRequests())
                        {
                            failoverRequest.ScheduleTimeout();
                        }

                        TriggerReconnectionAttempt();

                        listener?.OnConnectionInterrupted(failedUri);
                    }
                    else
                    {
                        listener?.OnConnectionFailure(cause);
                    }
                }
                else
                {
                    Tracer.Debug($"Ignoring duplicate provider failed event for provider: {provider}");
                }
            }
        }

        private void CheckClosed()
        {
            if (closed)
                throw new IOException("The Provider is already closed");
        }

        internal void AddFailoverRequest(FailoverRequest request)
        {
            lock (requests)
            {
                requests.Add(request);
            }
        }

        internal void RemoveFailoverRequest(FailoverRequest request)
        {
            lock (requests)
            {
                requests.Remove(request);
            }
        }

        internal IEnumerable<FailoverRequest> GetPendingRequests()
        {
            lock (requests)
            {
                return requests.ToArray();
            }
        }

        private class ReconnectControls
        {
            private readonly FailoverProvider failoverProvider;

            // Reconnection state tracking
            private volatile bool recoveryRequired;
            private long reconnectAttempts;
            private long nextReconnectDelay = -1;

            public ReconnectControls(FailoverProvider failoverProvider)
            {
                this.failoverProvider = failoverProvider;
            }

            public long ReconnectAttempts => reconnectAttempts;

            public async Task ScheduleReconnect(Func<Task> action)
            {
                // If no connection recovery required then we have never fully connected to a remote
                // so we proceed down the connect with one immediate connection attempt and then follow
                // on delayed attempts based on configuration.
                if (!recoveryRequired)
                {
                    if (ReconnectAttempts == 0)
                    {
                        Tracer.Debug("Initial connect attempt will be performed immediately");
                        await action().Await();;
                    }
                    else if (ReconnectAttempts == 1 && failoverProvider.InitialReconnectDelay > 0)
                    {
                        Tracer.Debug($"Delayed initial reconnect attempt will be in {failoverProvider.InitialReconnectDelay} milliseconds");
                        await Task.Delay(TimeSpan.FromMilliseconds(failoverProvider.InitialReconnectDelay)).Await();;
                        await action().Await();;
                    }
                    else
                    {
                        double delay = NextReconnectDelay();
                        Tracer.Debug($"Next reconnect attempt will be in {delay} milliseconds");
                        await Task.Delay(TimeSpan.FromMilliseconds(delay)).Await();;
                        await action().Await();;
                    }
                }
                else if (ReconnectAttempts == 0)
                {
                    if (failoverProvider.InitialReconnectDelay > 0)
                    {
                        Tracer.Debug($"Delayed initial reconnect attempt will be in {failoverProvider.InitialReconnectDelay} milliseconds");
                        await Task.Delay(TimeSpan.FromMilliseconds(failoverProvider.InitialReconnectDelay)).Await();
                        await action().Await();;
                    }
                    else
                    {
                        Tracer.Debug("Initial Reconnect attempt will be performed immediately");
                        await action().Await();;
                    }
                }
                else
                {
                    double delay = NextReconnectDelay();
                    Tracer.Debug($"Next reconnect attempt will be in {delay} milliseconds");
                    await Task.Delay(TimeSpan.FromMilliseconds(delay)).Await();
                    await action().Await();;
                }
            }

            private long NextReconnectDelay()
            {
                if (nextReconnectDelay == -1)
                {
                    nextReconnectDelay = failoverProvider.ReconnectDelay;
                }

                if (failoverProvider.UseReconnectBackOff && ReconnectAttempts > 1)
                {
                    // Exponential increment of reconnect delay.
                    nextReconnectDelay = (long) Math.Round(nextReconnectDelay * failoverProvider.ReconnectBackOffMultiplier);
                    if (nextReconnectDelay > failoverProvider.MaxReconnectDelay)
                    {
                        nextReconnectDelay = failoverProvider.MaxReconnectDelay;
                    }
                }

                return nextReconnectDelay;
            }

            public long RecordNextAttempt()
            {
                return ++reconnectAttempts;
            }

            public void ConnectionEstablished()
            {
                recoveryRequired = true;
                nextReconnectDelay = -1;
                reconnectAttempts = 0;
            }

            public bool IsReconnectAllowed(Exception cause)
            {
                // If a connection attempts fail due to Security errors than we abort
                // reconnection as there is a configuration issue and we want to avoid
                // a spinning reconnect cycle that can never complete.
                if (IsStoppageCause(cause))
                {
                    return false;
                }

                return !IsLimitExceeded();
            }

            private bool IsStoppageCause(Exception cause)
            {
                // TODO: Check if fail is due to Security errors
                return false;
            }

            public bool IsLimitExceeded()
            {
                int reconnectLimit = ReconnectAttemptLimit();
                if (reconnectLimit != UNDEFINED && reconnectAttempts >= reconnectLimit)
                {
                    return true;
                }

                return false;
            }

            private int ReconnectAttemptLimit()
            {
                int maxReconnectValue = failoverProvider.MaxReconnectAttempts;
                if (!recoveryRequired && failoverProvider.StartupMaxReconnectAttempts != UNDEFINED)
                {
                    // If this is the first connection attempt and a specific startup retry limit
                    // is configured then use it, otherwise use the main reconnect limit
                    maxReconnectValue = failoverProvider.StartupMaxReconnectAttempts;
                }

                return maxReconnectValue;
            }

            public bool IsRecoveryRequired()
            {
                return recoveryRequired;
            }
        }
    }
}