using System;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using xCommons.Extensions;
using xEventService.Configuration;
using xEventService.Constants;
using xEventService.Extensions;
using xExceptions.Constants;

namespace xEventService.Providers
{
    /// <summary>
    /// an Abstract Kafka Consumer Service ...
    /// </summary>
    public abstract class XKafkaConsumerServiceBase : BackgroundService
    {
        /// <summary>
        /// Consumed Topic Messages ...
        /// </summary>
        protected readonly string topic;

        /// <summary>
        /// Check is Paused ...
        /// </summary>
        private bool paused = false;

        /// <summary>
        /// Allowed Senders ...
        /// </summary>
        public readonly string[] allowedSenders;

        /// <summary>
        /// Allowed Actions ...
        /// </summary>
        public readonly string[] allowedActions;

        /// <summary>
        /// Kafka Consumer Class ...
        /// </summary>
        protected readonly IConsumer<Ignore, string> consumer;

        protected readonly XEventServiceConfiguration configuration;

        protected readonly ILogger<XKafkaConsumerServiceBase> logger;

        protected XKafkaConsumerServiceBase(
            string[] allowedActions,
            string[] allowedSenders,
            XEventServiceConfiguration configuration,
            ILogger<XKafkaConsumerServiceBase> logger,
            string topic = XEventServiceConstants.XDefaultTopic
        )
        {
            //
            this.topic = topic;
            this.logger = logger;
            this.configuration = configuration;

            //
            // Validate Actions ...
            if (!allowedActions
                .IsAllowedCollection())
            {
                XException.InvalidArgs.Throw();
            }
            this.allowedActions = [.. allowedActions];

            //
            // Validate Senders ...
            if (!allowedSenders
                .IsAllowedCollection())
            {
                XException.InvalidArgs.Throw();
            }
            this.allowedSenders = [.. allowedSenders];

            //
            // Prepare Kafka Consumer Configuration ...
            var config = new ConsumerConfig
            {
                EnableAutoCommit = false,
                BootstrapServers = configuration.Url,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                GroupId = XEventServiceConstants.XDefaultConsumerGroup,
            };
            consumer = new ConsumerBuilder<Ignore, string>(config).Build();
        }


        /// <summary>
        /// Pause Consuming ...
        /// </summary>
        /// <returns></returns>
        public bool Pause()
        {
            //
            var result = !paused;
            if (!result)
            {
                return result;
            }

            //
            paused = true;
            result = paused;
            return result;
        }

        /// <summary>
        /// Resume Consuming ...
        /// </summary>
        /// <returns></returns>
        public bool Resume()
        {
            //
            var result = paused;
            if (!result)
            {
                return result;
            }

            //
            paused = false;
            result = !paused;
            return result;
        }

        /// <summary>
        /// Default Action Execution for Background Services ...
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        protected override async Task ExecuteAsync(
            CancellationToken cancellationToken = default
        )
        {
            //
            // For Fix Blocking Synchronous ...
            await Task.Yield();
            using var cts = new CancellationTokenSource();

            // Subscribe to Topic ...
            consumer.Subscribe(topic);

            // Consuming ...
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    // Consuming Topic ...
                    await consumer.Consume(async (consumeResult, cancellationToken) =>
                    {
                        // Check Pausing ...
                        if (!paused)
                        {
                            // Logging Message Recieved ...
                            if (enableLogging)
                            {
                                // Encrypt Message ...
                                var expirationOffset = DateTimeOffset.UtcNow
                                    .AddMinutes(encryptionLifetimeInMinute);
                                var encryptedMessage = cryptographyService.Encrypt(
                                    text: consumeResult?.Message.Value ?? string.Empty,
                                    expirationOffset: expirationOffset
                                );
                                logger
                                    .LogInformation(
                                        "Consume Recieved Message from Kafka Topic: {Topic}, Offset: {Offset}, Partition: {Partition}, Message: {Message}",
                                        topic,
                                        consumeResult?.Offset,
                                        consumeResult?.Partition,
                                        encryptedMessage
                                );
                            }

                            // Produce Message Model ...
                            var model = consumeResult
                                .FillConsumeResult()
                                    ?? throw RpkKafkaExceptionHelper
                                        .GetInvalidKafkaConsumeResultException();

                            // Check Sender and Actions Owning ...
                            var isOwned = model.IsOwned(
                                allowedActions,
                                allowedSenders
                            );
                            if (isOwned)
                            {
                                // Notify Extended Classes for Consume Message ...
                                await ConsumeAsync(
                                    model,
                                    cancellationToken
                                );
                            }

                            // Commit Offset after Consuming ...
                            consumer?.Commit(consumeResult);
                        }
                    },
                    cts.Token);
                }
                catch (ConsumeException ex)
                {
                    // Log Exception ...
                    if (enableLogging)
                    {
                        logger.LogError(
                            ex,
                            "Kafka Message Consume Failed: {Error}",
                            ex.Error.Reason
                        );
                    }
                    await Task.Delay(1000, cancellationToken);
                }
                catch (Exception ex)
                {
                    // Log Exception ...
                    if (enableLogging)
                    {
                        logger.LogError(
                            ex,
                            "Exception: {Exc}",
                            ex.Message
                        );
                    }
                    await Task.Delay(1000, cancellationToken);
                }
            }
        }        

        /// <summary>
        /// Dispose Implementation ...
        /// </summary>
        public override void Dispose()
        {
            //
            consumer.Close();
            consumer.Dispose();
            GC.SuppressFinalize(this);
            base.Dispose();
        }
    }
}