// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

using System.Collections.Concurrent;

using Microsoft.Teams.Api;
using Microsoft.Teams.Api.Activities;
using Microsoft.Teams.Api.Entities;
using Microsoft.Teams.Apps.Plugins;
using Microsoft.Teams.Common.Logging;

using static Microsoft.Teams.Common.Extensions.TaskExtensions;

namespace Microsoft.Teams.Plugins.AspNetCore;

public partial class AspNetCorePlugin
{
    public class Stream : IStreamer
    {
        public bool Closed => _closedAt is not null;
        public int Count => _count;
        public int Sequence => _index;

        public required Func<IActivity, Task<IActivity>> Send { get; set; }
        public ILogger? Logger { get; set; }
        public event IStreamer.OnChunkHandler OnChunk = (_) => { };

        protected int _index = 1;
        protected string? _id;
        protected string _text = string.Empty;
        protected ChannelData _channelData = new();
        protected List<Attachment> _attachments = [];
        protected List<IEntity> _entities = [];
        protected ConcurrentQueue<IActivity> _queue = [];

        private DateTime? _closedAt;
        private int _count = 0;
        private MessageActivity? _result;
        private readonly SemaphoreSlim _lock = new(1, 1);
        private Timer? _timeout;

        public void Emit(MessageActivity activity)
        {
            if (_timeout != null)
            {
                _timeout.Dispose();
                _timeout = null;
            }

            _queue.Enqueue(activity);
            _timeout = new Timer(_ =>
            {
                _ = FlushSafe();
            }, null, 500, Timeout.Infinite);
        }

        public void Emit(TypingActivity activity)
        {
            if (_timeout != null)
            {
                _timeout.Dispose();
                _timeout = null;
            }

            _queue.Enqueue(activity);
            _timeout = new Timer(_ =>
            {
                _ = FlushSafe();
            }, null, 500, Timeout.Infinite);
        }

        public void Emit(string text)
        {
            Emit(new MessageActivity(text));
        }

        public void Update(string text)
        {
            Emit(new TypingActivity(text)
            {
                ChannelData = new()
                {
                    StreamType = StreamType.Informative
                }
            });
        }

        public async Task<MessageActivity?> Close(CancellationToken cancellationToken = default)
        {
            if (_index == 1 && _queue.Count == 0 && _lock.CurrentCount > 0) return null;
            if (_result is not null) return _result;
            // _lock.CurrentCount == 0 means Flush() is mid-await (queue drained but SendActivity calls
            // still pending). Wait it out so the final message doesn't race in-flight chunks.
            while (_id is null || _queue.Count > 0 || _lock.CurrentCount == 0)
            {
                await Task.Delay(50, cancellationToken).ConfigureAwait(false);
            }

            if (_text == string.Empty && _attachments.Count == 0) // when only informative updates are present
            {
                _text = "Streaming closed with no content";
            }

            var activity = new MessageActivity(_text)
                .AddAttachment(_attachments.ToArray());

            activity.WithId(_id);
            activity.WithData(_channelData);
            activity.AddEntity(_entities.ToArray());
            activity.AddStreamFinal();

            var res = await Retry(() => Send(activity)).ConfigureAwait(false);
            OnChunk(res);

            _result = activity;
            _closedAt = DateTime.Now;
            _index = 1;
            _id = null;
            _text = string.Empty;
            _attachments = [];
            _entities = [];
            _channelData = new();

            return (MessageActivity)res;
        }

        protected async Task Flush()
        {
            bool hasPendingState = _id is null && _count > 0 && _text != string.Empty;
            if (_queue.Count == 0 && !hasPendingState) return;

            await _lock.WaitAsync().ConfigureAwait(false);

            try
            {
                if (_timeout != null)
                {
                    _timeout.Dispose();
                    _timeout = null;
                }

                Queue<TypingActivity> informativeUpdates = new();
                var dequeued = 0;

                while (_queue.TryDequeue(out var activity))
                {
                    if (activity is MessageActivity message)
                    {
                        _text += message.Text;
                        _attachments.AddRange(message.Attachments ?? []);
                        _entities.AddRange(message.Entities ?? []);
                    }

                    if (activity.ChannelData is not null)
                    {
                        _channelData = _channelData.Merge(activity.ChannelData);
                    }

                    if (activity is TypingActivity typing && typing.ChannelData?.StreamType == StreamType.Informative && _text == string.Empty)
                    {
                        // If `_text` is not empty then it's possible that streaming has started.
                        // And so informative updates cannot be sent.
                        informativeUpdates.Enqueue(typing);
                    }

                    dequeued++;
                    _count++;
                }

                // Recalculate inside the lock to account for any concurrent changes.
                hasPendingState = _id is null && _count > 0 && _text != string.Empty;
                if (dequeued == 0 && !hasPendingState) return;

                // Send informative updates
                if (informativeUpdates.Count > 0)
                {
                    while (informativeUpdates.TryDequeue(out var typing))
                    {
                        await SendActivity(typing).ConfigureAwait(false);
                    }
                }

                // Send text chunk
                if (_text != string.Empty)
                {
                    var toSend = new TypingActivity(_text);
                    await SendActivity(toSend).ConfigureAwait(false);
                }

                if (_queue.Count > 0)
                {
                    _timeout = new Timer(_ =>
                    {
                        _ = FlushSafe();
                    }, null, 500, Timeout.Infinite);
                }

                async Task SendActivity(TypingActivity toSend)
                {
                    if (_id is not null)
                    {
                        toSend.WithId(_id);
                    }

                    toSend.AddStreamUpdate(_index);
                    var res = await Retry(() => Send(toSend)).ConfigureAwait(false);
                    OnChunk(res);
                    _id ??= res.Id;
                    _index++;
                }
            }
            finally
            {
                _lock.Release();
            }
        }

        private async Task FlushSafe()
        {
            try
            {
                await Flush().ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Logger?.Warn("Stream flush failed; will retry if there is pending state.", ex);
                // Reschedule a retry so Close() doesn't spin forever
                // waiting for _id to be set after a transient send failure.
                if (_queue.Count > 0 || _id is null && _count > 0)
                {
                    _timeout = new Timer(_ =>
                    {
                        _ = FlushSafe();
                    }, null, 1000, Timeout.Infinite);
                }
            }
        }
    }
}