using Ethanol.DataObjects;
using Ethanol.ContextBuilder.Schedulers;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Reactive;
using System.Reactive.Linq;
using System.Threading.Tasks.Dataflow;


namespace Ethanol.ContextBuilder.Reactive
{

    public static partial class ReactiveExtensionWindow
    {
        /// <summary>
        /// Creates a virtual window over an observable sequence of timestamped values.
        /// </summary>
        /// <typeparam name="T">The type of the values in the observable sequence.</typeparam>
        /// <param name="source">The source observable sequence.</param>
        /// <param name="windowSize">The size of the virtual window.</param>
        /// <param name="windowShift">The shift of the virtual window.</param>
        /// <returns>An observable sequence of observable sequences of timestamped values representing the virtual windows.</returns>
        public static IObservable<TimeRange<IObservable<Timestamped<T>>>> VirtualWindow<T>(
            this IObservable<Timestamped<T>> source,
            TimeSpan windowSize,
            TimeSpan windowShift)
        {
            var virtualScheduler = new VirtualFlowTimeScheduler();
            var firstFlow = true;

            DateTimeOffset GetWindowStart(DateTimeOffset timestamp)
            {
                var timeOfDayTicks = timestamp.TimeOfDay.Ticks;
                var windowTicks = windowSize.Ticks % TimeSpan.FromDays(1).Ticks;   //this is for sure that we are within a single day
                var shiftTicks = timeOfDayTicks % windowTicks;
                var windowStart = timestamp - TimeSpan.FromTicks(shiftTicks);
                return windowStart;
            }

            void ShiftTime(Timestamped<T> flow)
            {
                if (firstFlow)
                {
                    var windowStart = GetWindowStart(flow.Timestamp);
                    virtualScheduler.SkipTo(windowStart);
                    firstFlow = false;
                }
                if (flow.Timestamp > virtualScheduler.Clock)
                {
                    virtualScheduler.AdvanceTo(flow.Timestamp);
                }
            }

            return System.Reactive.Linq.Observable.Create<TimeRange<IObservable<Timestamped<T>>>>(observer =>
            {
                return source
                    .Do(ShiftTime).Window(windowSize, windowShift, virtualScheduler).Timestamp(virtualScheduler)
                    .Select(w => new TimeRange<IObservable<Timestamped<T>>>(w.Value, w.Timestamp, w.Timestamp + windowSize))
                    .Subscribe(observer);
            });
        }

        /// <summary>
        /// Groups the elements of the source observable sequence into windows based on a specified time window size,
        /// and then groups the elements within each window by a key selector function.
        /// </summary>
        /// <typeparam name="TKey">The type of the key returned by the key selector function.</typeparam>
        /// <typeparam name="TValue">The type of the elements in the source observable sequence.</typeparam>
        /// <param name="source">The source observable sequence of timestamped values.</param>
        /// <param name="keySelector">A function to extract the key for each element in the source observable sequence.</param>
        /// <param name="windowSize">The size of the time windows.</param>
        /// <returns>An observable sequence of grouped observables representing the windows.</returns>
        public static IObservable<TimeRange<IGroupedObservable<TKey,TValue>>> WindowGroupBy<TKey, TValue>(this IObservable<Timestamped<TValue>> source, 
            Func<Timestamped<TValue>, TKey> keySelector,
            TimeSpan windowSize)
        {
            var virtualScheduler = new VirtualFlowTimeScheduler();
            var firstFlow = true;

            DateTimeOffset GetWindowStart(DateTimeOffset timestamp)
            {
                var timeOfDayTicks = timestamp.TimeOfDay.Ticks;
                var windowTicks = windowSize.Ticks % TimeSpan.FromDays(1).Ticks;   //this is for sure that we are within a single day
                var shiftTicks = timeOfDayTicks % windowTicks;
                var windowStart = timestamp - TimeSpan.FromTicks(shiftTicks);
                return windowStart;
            }

            void ShiftTime(Timestamped<TValue> flow)
            {
                if (firstFlow)
                {
                    var windowStart = GetWindowStart(flow.Timestamp);
                    virtualScheduler.SkipTo(windowStart);
                    firstFlow = false;
                }
                if (flow.Timestamp > virtualScheduler.Clock)
                {
                    virtualScheduler.AdvanceTo(flow.Timestamp);
                }
            }
  
            TimeRange<IGroupedObservable<TKey, TValue>> GetTimeRangeGroup(Timestamped<IGroupedObservable<TKey, TValue>> item)
            {
                var start = GetWindowStart(item.Timestamp);
                var end = start + windowSize;
                return new TimeRange<IGroupedObservable<TKey, TValue>>(item.Value, start, end);
            }

            return Observable.Create<TimeRange<IGroupedObservable<TKey, TValue>>>(observer =>
            {
                var t = source.Do(ShiftTime).GroupByUntil(keySelector, item=>item.Value, grp => Observable.Timer( GetWindowStart(virtualScheduler.Now) + windowSize, virtualScheduler)).Timestamp(virtualScheduler);
                var s = t.Select(GetTimeRangeGroup);
                return s.Subscribe(observer);
            });
        }

        /// <summary>
        /// Aggregates the elements of an observable sequence using given key selector function.
        /// Creates Observable events with the given time window.
        /// </summary>
        /// <typeparam name="T">The type of the elements in the source observable sequence.</typeparam>
        /// <typeparam name="R">The type of the result.</typeparam>
        /// <typeparam name="K">The type of the key.</typeparam>
        /// <typeparam name="V">The type of the value.</typeparam>
        /// <param name="source">The source observable sequence.</param>
        /// <param name="windowStartTime">The start time of the time window.</param>
        /// <param name="windowEndTime">The end time of the time window.</param>
        /// <param name="keyValSelector">A function that maps each element to an array of key-value pairs.</param>
        /// <param name="resultSelector">A function to aggregate the values for each key into a result.</param>
        /// <param name="emptyResultSelector">A function to generate an empty result.</param>
        /// <returns>An observable sequence of aggregated results within the specified time window.</returns>
        public static IObservable<TimeRange<R>> Aggregate<T, R, K, V>(
                this IObservable<Timestamped<T>> source,
                DateTimeOffset windowStartTime, DateTimeOffset windowEndTime,
                Func<Timestamped<T>, KeyValuePair<K, V?>[]> keyValSelector,
                Func<KeyValuePair<K, V?[]>, R> resultSelector,
                Func<R> emptyResultSelector
        )
        {
            return source
                    .SelectMany(keyValSelector)
                    .GroupByAggregate(k => k.Key, v => v.Value, g => new TimeRange<R>(resultSelector(g), windowStartTime, windowEndTime))
                    .Append(new TimeRange<R>(emptyResultSelector(), windowStartTime, windowEndTime));

        }

        /// <summary>
        /// Aggregates the IP flows within a specified time window and returns an observable sequence of observable events containing the aggregated IP host contexts.
        /// </summary>
        /// <param name="flows">The observable sequence of timestamped IP flows.</param>
        /// <param name="windowStartTime">The start time of the window.</param>
        /// <param name="windowEndTime">The end time of the window.</param>
        /// <returns>An observable sequence of observable events containing the aggregated IP host contexts.</returns>
        public static IObservable<TimeRange<IpHostContext>> AggregateIpContexts(this IObservable<Timestamped<IpFlow>> flows, DateTimeOffset windowStartTime, DateTimeOffset windowEndTime)
        {
            return flows.Aggregate(
                windowStartTime, windowEndTime,
                bihostFlowKeySelector,
                g => new IpHostContext { HostAddress = g.Key, Flows = g.Value.Where(f => f != null).Select(f => f!).ToArray() },
                () => new IpHostContext { HostAddress = IPAddress.Any, Flows = Array.Empty<IpFlow>() }
            );
        }

        /// <summary>
        /// Duplicates the given IP flow by creating two key-value pairs, one with the source address and one with the destination address.
        /// </summary>
        /// <param name="flow">The IP flow to duplicate.</param>
        /// <returns>An array of key-value pairs, where the key is the IP address and the value is the duplicated IP flow.</returns>
        static KeyValuePair<IPAddress, IpFlow?>[] bihostFlowKeySelector(Timestamped<IpFlow> flow)
        {
            return new[] {
                    new KeyValuePair<IPAddress, IpFlow?>(flow.Value?.SourceAddress ?? IPAddress.None, flow.Value),
                    new KeyValuePair<IPAddress, IpFlow?>(flow.Value?.DestinationAddress ?? IPAddress.None, flow.Value) };
        }
    }
}