//******************************************************************************************************
//  SubscriberHub.cs - Gbtc
//
//  Copyright © 2018, Grid Protection Alliance.  All Rights Reserved.
//
//  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See
//  the NOTICE file distributed with this work for additional information regarding copyright ownership.
//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may not use this
//  file except in compliance with the License. You may obtain a copy of the License at:
//
//      http://opensource.org/licenses/MIT
//
//  Unless agreed to in writing, the subject software distributed under the License is distributed on an
//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the
//  License for the specific language governing permissions and limitations.
//
//  Code Modification History:
//  ----------------------------------------------------------------------------------------------------
//  07/12/2018 - Stephen C. Wills
//       Generated original version of source code.
//
//******************************************************************************************************

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Threading.Tasks;
using GSF.Collections;
using GSF.Configuration;
using GSF.TimeSeries;
using GSF.TimeSeries.Transport;
using GSF.Web.Security;
using Microsoft.AspNet.SignalR;
using Newtonsoft.Json.Linq;

namespace GSF.Web.Shared
{
    /// <summary>
    /// SignalR hub that exposes server-side functions for the subscriber API.
    /// </summary>
    [AuthorizeHubRole]
    public class SubscriberHub : Hub
    {
        #region [ Members ]

        // Nested Types

        private sealed class Connection : IDisposable
        {
            public readonly Dictionary<string, Subscriber> SubscriberLookup;
            private bool m_disposed;

            public Connection()
            {
                SubscriberLookup = new Dictionary<string, Subscriber>();
            }

            public void Dispose()
            {
                if (m_disposed)
                    return;

                try
                {
                    foreach (Subscriber subscriptionState in SubscriberLookup.Values)
                        subscriptionState.Dispose();
                }
                finally
                {
                    m_disposed = true;
                }
            }
        }

        private sealed class Subscriber : IDisposable
        {
            public DataSubscriber DataSubscriber;
            public DataSet Metadata;

            private Dictionary<Guid, Tuple<string, string>> m_formats;
            private bool m_disposed;

            public void SetFormat(Guid signalID, string format, string dataType)
            {
                if (m_formats is null)
                    m_formats = new Dictionary<Guid, Tuple<string, string>>();

                m_formats[signalID] = new Tuple<string, string>(format, dataType);
            }

            public bool TryGetFormat(Guid signalID, out string format, out string dataType)
            {
                format = null;
                dataType = null;

                if (m_formats is null)
                    return false;

                if (m_formats.TryGetValue(signalID, out Tuple<string, string> tuple))
                {
                    format = tuple.Item1;
                    dataType = tuple.Item2;
                    return true;
                }

                return false;
            }

            public void RemoveFormat(Guid signalID)
            {
                if (m_formats is null || !m_formats.ContainsKey(signalID))
                    return;

                if (m_formats.Remove(signalID) && m_formats.Count == 0)
                    m_formats = null;
            }

            public void Dispose()
            {
                if (m_disposed)
                    return;

                try
                {
                    DataSubscriber?.Dispose();
                    Metadata?.Dispose();
                }
                finally
                {
                    m_disposed = true;
                }
            }
        }

        #endregion

        #region [ Methods ]

        /// <summary>
        /// Initiates the subscriber connection.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber to be connected.</param>
        public void Connect(string subscriberID)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);
            subscriber.DataSubscriber.Start();
        }

        /// <summary>
        /// Sends a command to the publisher.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        /// <param name="commandCode">The command to be sent.</param>
        /// <param name="message">The message to be sent to the publisher.</param>
        public void SendCommand(string subscriberID, ServerCommand commandCode, string message)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);
            subscriber.DataSubscriber.SendServerCommand(commandCode, message);
        }

        /// <summary>
        /// Filters metadata and returns the result.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        /// <param name="tableName">The metadata table from which to return rows.</param>
        /// <param name="filter">The filter to apply to the metadata table.</param>
        /// <param name="sortField">The field by which to sort the result set.</param>
        /// <param name="takeCount">The maximum number of records to be returned from the result set.</param>
        /// <returns>The result of the metadata query.</returns>
        public IEnumerable<object> GetMetadata(string subscriberID, string tableName, string filter, string sortField, int takeCount)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);
            DataSet metadata = subscriber.Metadata;

            if (metadata is null)
                return null;

            if (string.IsNullOrEmpty(tableName) || !metadata.Tables.Contains(tableName))
                return null;

            DataTable table = metadata.Tables[tableName];

            IEnumerable<DataRow> rows = string.IsNullOrEmpty(filter) ? table.Select() : table.Select(filter);

            if (!string.IsNullOrEmpty(sortField) && table.Columns.Contains(sortField))
                rows = rows.OrderBy(row => row[sortField]);

            if (takeCount > 0)
                rows = rows.Take(takeCount);

            return rows.Select(FromDataRow);
        }

        /// <summary>
        /// Defines a string format to apply for a given measurement.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        /// <param name="signalID">Measurement signal ID to which to apply the format.</param>
        /// <param name="format">String format to apply, e.g., "{0:N3} seconds".</param>
        /// <param name="dataType">Fully qualified data type for measurement, defaults to "System.Double" if <c>null</c>.</param>
        /// <remarks>
        /// <para>
        /// Set <paramref name="format"/> to <c>null</c> to remove any existing formatting.
        /// </para>
        /// <para>
        /// Conversion of double-precision measurement floating point value to specified
        /// <paramref name="dataType"/> will be attempted when type is provided.
        /// </para>
        /// </remarks>
        public void SetMeasurementFormat(string subscriberID, Guid signalID, string format, string dataType)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);

            if (string.IsNullOrWhiteSpace(format))
                subscriber.RemoveFormat(signalID);
            else
                subscriber.SetFormat(signalID, format, dataType);
        }

        /// <summary>
        /// Defines string formats to apply for the collection of format records.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        /// <param name="formatRecords">Collection of format records.</param>
        /// <remarks>
        /// This is simply a bulk operation for <see cref="SetMeasurementFormat"/>.
        /// Each format record should be a Json object similar to the following:
        /// <code>
        /// {
        ///     "signalID":"4B1DEE7C-72EC-41EA-AAF3-7E8094355740",
        ///     "format":"{0:N3} seconds",
        ///     "dataType":"System.Double"
        /// }
        /// </code>
        /// Where
        ///  - "signalID" is the measurement signal ID to which to apply the format
        ///  - "format" is the string format to apply, and
        ///  - "dataType" is the fully qualified data type for measurement
        /// </remarks>
        public void SetMeasurementFormats(string subscriberID, IEnumerable<dynamic> formatRecords)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);

            foreach (dynamic formatRecord in formatRecords)
            {
                Guid signalID = formatRecord.signalID;
                string format = formatRecord.format;
                string dataType = formatRecord.dataType;

                if (string.IsNullOrWhiteSpace(format))
                    subscriber.RemoveFormat(signalID);
                else
                    subscriber.SetFormat(signalID, format, dataType);
            }
        }

        /// <summary>
        /// Subscribes to the internal data publisher.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        /// <param name="data">Data from the client describing the subscription.</param>
        public void Subscribe(string subscriberID, JObject data)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);
            SubscriptionInfo subscriptionInfo = ToSubscriptionInfo(subscriberID, data);
            subscriber.DataSubscriber.Subscribe(subscriptionInfo);
        }

        /// <summary>
        /// Unsubscribes from the publisher.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        public void Unsubscribe(string subscriberID)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);
            subscriber.DataSubscriber.Unsubscribe();
        }

        /// <summary>
        /// Disconnects from the publisher.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        public void Disconnect(string subscriberID)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);
            subscriber.DataSubscriber.Stop();
        }

        /// <summary>
        /// Closes the connection and releases all resources retained by the subscriber with this ID.
        /// </summary>
        /// <param name="subscriberID">The ID of the subscriber.</param>
        public void Dispose(string subscriberID)
        {
            string connectionID = Context.ConnectionId;

            if (s_connectionLookup.TryGetValue(connectionID, out Connection connection) && connection.SubscriberLookup.TryGetValue(subscriberID, out Subscriber subscriber))
            {
                connection.SubscriberLookup.Remove(subscriberID);
                subscriber.Dispose();
            }
        }

        /// <summary>
        /// Handles when a client disconnects from the hub.
        /// </summary>
        /// <param name="stopCalled">Indicates whether the client called stop.</param>
        /// <returns>The task to handle the disconnect.</returns>
        public override Task OnDisconnected(bool stopCalled)
        {
            return Task.Run(() =>
            {
                string connectionID = Context.ConnectionId;

                if (s_connectionLookup.TryRemove(connectionID, out Connection connection))
                    connection.Dispose();
            });
        }

        private Subscriber GetOrCreate(string subscriberID)
        {
            string connectionID = Context.ConnectionId;
            Connection connection = s_connectionLookup.GetOrAdd(connectionID, _ => new Connection());

            return connection.SubscriberLookup.GetOrAdd(subscriberID, _ =>
            {
                DataSubscriber dataSubscriber = new DataSubscriber();
                Subscriber subscriber = new Subscriber { DataSubscriber = dataSubscriber };

                dataSubscriber.ConnectionEstablished += (_, _) => Subscriber_ConnectionEstablished(connectionID, subscriberID);
                dataSubscriber.ConnectionTerminated += (_, _) => Subscriber_ConnectionTerminated(connectionID, subscriberID);
                dataSubscriber.MetaDataReceived += (_, args) => Subscriber_MetadataReceived(connectionID, subscriberID, subscriber, args.Argument);
                dataSubscriber.ConfigurationChanged += (_, _) => Subscriber_ConfigurationChanged(connectionID, subscriberID);
                dataSubscriber.NewMeasurements += (_, args) => Subscriber_NewMeasurements(connectionID, subscriberID, args.Argument);
                dataSubscriber.StatusMessage += (_, args) => Subscriber_StatusMessage(connectionID, subscriberID, args.Argument);
                dataSubscriber.ProcessException += (_, args) => Subscriber_ProcessException(connectionID, subscriberID, args.Argument);
                dataSubscriber.ConnectionString = GetInternalPublisherConnectionString();
                dataSubscriber.CompressionModes = CompressionModes.TSSC | CompressionModes.GZip;
                dataSubscriber.AutoSynchronizeMetadata = false;
                dataSubscriber.ReceiveInternalMetadata = true;
                dataSubscriber.ReceiveExternalMetadata = true;

                dataSubscriber.Initialize();

                return subscriber;
            });
        }

        private SubscriptionInfo ToSubscriptionInfo(string subscriberID, JObject obj)
        {
            dynamic info = obj;
            bool synchronized = info.Synchronized ?? false;

            IEnumerable<dynamic> formatRecords = info.FormatRecords;

            if (!(formatRecords is null))
                SetMeasurementFormats(subscriberID, formatRecords);

            if (synchronized)
                return ToSynchronizedSubscriptionInfo(obj);

            return ToUnsynchronizedSubscriptionInfo(obj);
        }

        private UnsynchronizedSubscriptionInfo ToUnsynchronizedSubscriptionInfo(JObject obj)
        {
            dynamic info = obj;

            if (info.throttled is null)
                info.throttled = info.Throttled ?? false;

            return obj.ToObject<UnsynchronizedSubscriptionInfo>();
        }

        private void Subscriber_ConnectionEstablished(string connectionID, string subscriberID) => 
            Clients.Client(connectionID).ConnectionEstablished(subscriberID);

        private void Subscriber_ConnectionTerminated(string connectionID, string subscriberID) => 
            Clients.Client(connectionID).ConnectionTerminated(connectionID, subscriberID);

        private void Subscriber_NewMeasurements(string connectionID, string subscriberID, ICollection<IMeasurement> measurements)
        {
            Subscriber subscriber = GetOrCreate(subscriberID);

            object data = measurements
                .Select(measurement => ToJsonMeasurement(subscriber, measurement))
                .ToArray();

            Clients.Client(connectionID).NewMeasurements(subscriberID, data);
        }

        private void Subscriber_MetadataReceived(string connectionID, string subscriberID, Subscriber subscriber, DataSet metadata)
        {
            subscriber.Metadata = metadata;
            Clients.Client(connectionID).MetadataReceived(subscriberID);
        }

        private void Subscriber_ConfigurationChanged(string connectionID, string subscriberID) => 
            Clients.Client(connectionID).ConfigurationChanged(subscriberID);

        private void Subscriber_StatusMessage(string connectionID, string subscriberID, string message) => 
            Clients.Client(connectionID).StatusMessage(subscriberID, message);

        private void Subscriber_ProcessException(string connectionID, string subscriberID, Exception ex) => 
            Clients.Client(connectionID).ProcessException(subscriberID, ex);

        #endregion

        #region [ Static ]

        // Static Fields
        private static readonly ConcurrentDictionary<string, Connection> s_connectionLookup;

        // Static Constructor
        static SubscriberHub()
        {
            s_connectionLookup = new ConcurrentDictionary<string, Connection>();
        }

        // Static Methods
        private static string GetInternalPublisherConnectionString()
        {
            ConfigurationFile configurationFile = ConfigurationFile.Current;
            CategorizedSettingsElementCollection internalDataPublisher = configurationFile.Settings["internaldatapublisher"];
            string configurationString = internalDataPublisher["ConfigurationString"]?.Value ?? "";
            Dictionary<string, string> settings = configurationString.ParseKeyValuePairs();

            if (!settings.TryGetValue("port", out string portSetting) || !int.TryParse(portSetting, out int port))
                port = 6165;

            if (settings.TryGetValue("interface", out string interfaceSetting))
                interfaceSetting = $"Interface={interfaceSetting}";
            else
                interfaceSetting = string.Empty;

            return $"Server=localhost:{port};{interfaceSetting};BypassStatistics=true";
        }

        private static object ConvertValueToType(double value, string dataType)
        {
            if (string.IsNullOrWhiteSpace(dataType))
                dataType = "System.Double";

            try
            {
                return dataType switch
                {
                    "System.Double"   => value,
                    "System.DateTime" => new DateTime((long)value),
                    "GSF.UnixTimeTag" => new UnixTimeTag((decimal)value),
                    _                 => Convert.ChangeType(value, Type.GetType(dataType) ?? typeof(double))
                };
            }
            catch
            {
                return value;
            }
        }

        private static SynchronizedSubscriptionInfo ToSynchronizedSubscriptionInfo(JObject obj)
        {
            dynamic info = obj;

            if (info.remotelySynchronized is null)
                info.remotelySynchronized = info.RemotelySynchronized ?? false;

            if (info.framesPerSecond is null)
                info.framesPerSecond = info.FramesPerSecond ?? 30;

            return obj.ToObject<SynchronizedSubscriptionInfo>();
        }

        private static object FromDataRow(DataRow dataRow)
        {
            JObject obj = new JObject();

            foreach (DataColumn dataColumn in dataRow.Table.Columns)
                obj[dataColumn.ColumnName.ToLowerInvariant()] = JToken.FromObject(dataRow[dataColumn]);

            return obj;
        }

        private static object ToJsonMeasurement(Subscriber subscriber, IMeasurement measurement)
        {
            Guid signalID = measurement.ID;
            double value = measurement.AdjustedValue;
            dynamic obj = new JObject();

            // Custom property made lowercase to match lowercase field
            // definitions to make Javascript code less confusing
            obj.signalid = signalID;

            if (subscriber.TryGetFormat(signalID, out string format, out string dataType))
                obj.value = string.Format(format, ConvertValueToType(value, dataType));
            else
                obj.value = value;

            obj.timestamp = (measurement.Timestamp - UnixTimeTag.BaseTicks).ToMilliseconds();

            return obj;
        }

        #endregion
    }
}