qdr.fnd/qdr.fnd.core.pqueue/PsQueue.cs

using System.Collections.Concurrent;
using qdr.fnd.core.pqueue.Configuration;
using qdr.fnd.core.pqueue.Enums;
using qdr.fnd.core.pqueue.Exceptions;
using qdr.fnd.core.pqueue.Extensions;
using qdr.fnd.core.pqueue.Process;
using qdr.fnd.core.pqueue.Storages;
using Quadarax.Foundation.Core.Data;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Thread;
using Quadarax.Foundation.Core.Value.Extensions;
using Quadarax.Foundation.Core.Value.Generators;

namespace qdr.fnd.core.pqueue
{
    public class PsQueue : LoopWorker, IPsQueue
    {
        #region *** Constants ***
        private const string CS_DEF_QUEUE_NAME = "PSQUEUE";
        public const int CN_PRIO_SYSTEM = 0;
        public const int CN_PRIO_IMMEDIATE = 2;
        public const int CN_PRIO_CUSTOM = 5;
        #endregion

        #region *** Properties ***
        public PsQueueConfiguration Configuration { get;}

        public event PsQueueStatusChangedEventHandler? ProcessItemStatusChanged;
        #endregion

        #region *** Private Fields ***
        private TinyHash _hashGenerator;
        protected IPsQueueStorage _storage;
        private List<Tuple<PsStatusEnum,PsStatusEnum[]>> _statusTransitions = new List<Tuple<PsStatusEnum, PsStatusEnum[]>>();

        private ConcurrentDictionary<string, IPsProcessProvider> _workingItems = new ConcurrentDictionary<string, IPsProcessProvider>();
        #endregion

        #region *** Constructors ***

        public PsQueue(string queueName, PsQueueConfiguration configuration,IPsQueueStorage storageProvider, ILogger logger) : base(queueName,null)
        {
            _hashGenerator = new TinyHash(5);
            if (string.IsNullOrEmpty(queueName))
                queueName =CS_DEF_QUEUE_NAME + _hashGenerator.NewId();

            Configuration = configuration ?? throw new ArgumentNullException(nameof(configuration));
            _storage = storageProvider ?? throw new ArgumentNullException(nameof(storageProvider));

            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.New,new[]
                {
                    PsStatusEnum.Pending
                }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Pending,new[]
            {
                PsStatusEnum.Started,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Started,new[]
            {
                PsStatusEnum.Pending,
                PsStatusEnum.Paused,
                PsStatusEnum.Stopped,
                PsStatusEnum.DoneOk,
                PsStatusEnum.DoneFailed,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Paused,new[]
            {
                PsStatusEnum.Started,
                PsStatusEnum.Stopped,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Stopped,new[]
            {
                PsStatusEnum.DoneFailed,
                PsStatusEnum.Expired
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.DoneOk,new[]
            {
                PsStatusEnum.Deleted
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.DoneFailed,new[]
            {
                PsStatusEnum.Deleted
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Expired,new[]
            {
                PsStatusEnum.Deleted
            }));
            _statusTransitions.Add(new Tuple<PsStatusEnum, PsStatusEnum[]>(PsStatusEnum.Deleted,Array.Empty<PsStatusEnum>()));

        }

        #endregion

        #region *** Public Operations ***
       
        public IQueryable<IPsQueueItem> Query(string? owner, string? ownerRef, params Tuple<bool,PsStatusEnum>[] status)
        {
            throw new NotImplementedException();
        }

        /// <inheritdoc/>
        public string EnQueueItem(IPsQueueItemDescriptor item)
        {
            ValidateItem(item);

            Log(LogSeverityEnum.Debug,$"{Name} - Item {item} begins to enqueue.");
            var itemId = _storage.CreateItem(item.Owner, item.OwnerRef, item.ProcessProviderClass, item.AtteptsInitial, null);
            foreach (var argument in item.Arguments)
            {
                if (argument.ValueType.FullName == null) throw new InvalidOperationException($"Cannot obtain fullName of type '{argument.ValueType.Name}'");
                
                _storage.CreateItemAttribute(itemId, argument.Name, argument.Value, argument.ValueType.FullName, argument.IsOutput);
                Log(LogSeverityEnum.Debug,$"{Name} - Attribute {argument.Name} was append to item [{itemId}]");
             }
            var newItem = GetItemInternal(itemId);
            SetItemStatusInternal(newItem, PsStatusEnum.Pending, true, "Item was enqueued.");
            Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was enqueued as {newItem}");
            return itemId;
        }

        /// <inheritdoc/>
        public IPsQueueItem GetQueueItem(string itemId)
        {
            ValidateItemId(itemId);

            Log(LogSeverityEnum.Debug,$"{Name} - Item [{itemId}] begins to get.");
            var item = GetItemInternal(itemId);
            Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was fetched as {item}");
            return item;
        }

        /// <inheritdoc/>
        public void ForceStartQueueItem(string itemId, string emitter)
        {
            ValidateItemId(itemId);
            ValidateEmitter(emitter);

            Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for force start by '{emitter}'.");
            var item = _storage.GetItem(itemId);
            EnsureItemStateExpired(item);

            if (item.Status.In(PsStatusEnum.Started, PsStatusEnum.Stopped, PsStatusEnum.Expired))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForForceStart, item.Status);

            _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom));
            _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE);
            SetItemStatusInternal(item, PsStatusEnum.Pending, true, $"Item will reset due force start by emitter '{emitter}'.");
            Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for force start by {emitter}.");
        }

        /// <inheritdoc/>
        public void ForceStopQueueItem(string itemId, string emitter)
        {
            ValidateItemId(itemId);
            ValidateEmitter(emitter);

            Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for stop start by '{emitter}'.");
            var item = _storage.GetItem(itemId);
            EnsureItemStateExpired(item);

            if (item.Status.In(PsStatusEnum.Expired, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForForceStop, item.Status);

            _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom));
            _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE);
            SetItemStatusInternal(item, PsStatusEnum.Stopped, true, $"Item will reset due force stop by emitter '{emitter}'.");
            Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for force stop by {emitter}.");

        }

        /// <inheritdoc/>
        public void PauseQueueItem(string itemId, string emitter)
        {
            ValidateItemId(itemId);
            ValidateEmitter(emitter);

            Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for pause by '{emitter}'.");
            var item = _storage.GetItem(itemId);
            EnsureItemStateExpired(item);

            if (item.Status.In(PsStatusEnum.Expired, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForPause, item.Status);

            _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom));
            _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE);
            SetItemStatusInternal(item, PsStatusEnum.Paused, true, $"Item will reset due pause by emitter '{emitter}'.");
            Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for pause by {emitter}.");

        }

        public void StartQueueItem(string itemId, string emmiter, bool isImmediate)
        {
            throw new NotImplementedException();
        }

        /// <inheritdoc/>
        public IQueryable<IPsQueueItem> QueryQueueItems(IPaging paging, string? owner, string? ownerRef, bool includeTransient = false, params PsStatusEnum[] statuses)
        {
            if (paging == null)
                throw new ArgumentNullException(nameof(paging));
            var statusesForQuery = new List<Tuple<bool, PsStatusEnum>>();
            foreach (var status in statuses)
            {
                statusesForQuery.Add(new Tuple<bool, PsStatusEnum>(false, status));
                if (includeTransient)
                    statusesForQuery.Add(new Tuple<bool, PsStatusEnum>(true, status));
            }
            return _storage.Query(owner, ownerRef, statusesForQuery.ToArray());

        }

        public void Start()
        {
            base.Start();
        }

        public void Stop()
        {
            base.Stop();
        }
        #endregion


        #region *** Private Operations ***

        #region Overrides of LoopWorker

        /// <inheritdoc />
        protected override Task DoIteration(IWorkerContext? context)
        {
            // fetch items to immediate stop - frees workers
            var itemsToStop = Query(null, null,
                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Stopped),new(true, PsStatusEnum.Paused)  })
                .OrderBy(x=>x.Priority).ThenBy(x=>x.ValidFrom);


            foreach(var item in itemsToStop){
                
                var process = _workingItems.TryGetValue(item.Id, out var processProvider) ? processProvider : null;
                if (process == null)
                    Log(LogSeverityEnum.Warn, $"{Name} - Cannot find process provider in working cache for item [{item.Id}]. Skiped.");
                if (process != null)
                { 
                    process.Stop();
                    SetItemStatusInternal(item, item.Status, false, "Item was stopped by schedule.");
                    _workingItems.TryRemove(item.Id, out _);
                }
            }

            // fetch working, stopped, paused items which currently expired - frees workers
            var itemsToExpire = Query(null, null,
                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Stopped),new(true, PsStatusEnum.Paused), new(true, PsStatusEnum.Started)  })
                .Where(x=>x.IsValiditySpanExpired());
            foreach(var item in itemsToExpire)
            {
                EnsureItemStateExpired(item);
            }

            // fetch items to immediate start
            var workersLeft = Configuration.MaxConcurrentProcesses - _workingItems.Count;
            if (workersLeft>0)
            {
                // fetch 
                var itemsToProcess = Query(null, null,
                        new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Started) })
                    .Where(x => !x.IsValiditySpanExpired()).Take(workersLeft);

                foreach(var item in itemsToProcess)
                {
                    if (item.AttemptsLeft <= 0)
                    {
                        SetItemStatusInternal(item, PsStatusEnum.DoneFailed, true, "Item will be marked as failed due no attempts left.");
                        SetItemStatusInternal(item, PsStatusEnum.DoneFailed, false, "Item will be marked as failed due no attempts left.");
                        continue;
                    }
                    _storage.UpdateItemAttempts(item.Id, item.AttemptsLeft - 1);
                    var provider = ResolveProvider(item.ProcessProviderClass, item.Arguments);
                    _storage.UpdateItemStarted(item.Id, DateTime.Now);
                    _storage.UpdateItemFinished(item.Id, null);
                    _workingItems.TryAdd(item.Id, provider);
                    provider.Start();
                    SetItemStatusInternal(item, PsStatusEnum.Started, false, "Item was started by schedule.");
                }
            }

            // fetch items to close expired
            var itemsToClose = Query(null, null,
                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Expired) });
            foreach (var item in itemsToClose)
            {
                SetItemStatusInternal(item, PsStatusEnum.Expired, false, "Item will be expired by schedule.");
            }

            // fetch pending items to initiate to expired
            var itemsPendingToExpire = Query(null, null,
                    new Tuple<bool, PsStatusEnum>[] { new(true, PsStatusEnum.Pending), new(false, PsStatusEnum.Pending) });
            foreach(var item in itemsPendingToExpire)
            {
                SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by schedule.");
            }


            return base.DoIteration(context);
        }

        #endregion

        protected void ValidateItemId(string itemId)
        {
            if (string.IsNullOrEmpty(itemId))
                throw new ArgumentNullException(nameof(itemId));
        }
        protected void ValidateEmitter(string emitter)
        {
            if (string.IsNullOrEmpty(emitter))
                throw new ArgumentNullException(nameof(emitter));
        }
        protected void ValidateItem(IPsQueueItemDescriptor item)
        {
            if (item is null)
                throw new ArgumentNullException(nameof(item));

            if (string.IsNullOrEmpty(item.Owner))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.OwnerRequired);

            if (string.IsNullOrEmpty(item.OwnerRef))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.OwnerRefRequired);

            var provider =  ResolveProvider(item.ProcessProviderClass, item.Arguments);
            provider.Validate();

            if (item.AtteptsInitial <= 0)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InitialAttemptMustBePositive);
                
        }

        protected void SetItemStatusInternal(IPsQueueItem item, PsStatusEnum? newStatus, bool isTransient, string message)
        {
            var oldStatus = item.Status;
            var oldIsTransient = item.IsStatusTransient;
            
            if (!isTransient && newStatus == null)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusTransientCombination);
           
            if (newStatus != null)
            {
                if (!_statusTransitions.First(x => x.Item1 == oldStatus).Item2.Contains(newStatus.Value))
                    throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusTransition, oldStatus,newStatus ?? oldStatus, string.Join(",", _statusTransitions.First(x => x.Item1 == oldStatus).Item2.Select(x=>x.ToString())));
            }

            _storage.AppendStatusJournal(item.Id, newStatus ?? oldStatus, isTransient, message);
            _storage.UpdateItemStatus(item.Id, newStatus, isTransient);
            ((PsQueueItem)item).SetStatus(newStatus, isTransient, message);
            ProcessItemStatusChanged?.Invoke(this, new PsQueueStatusChangedEventArgs(item.Id) { CurrentStatus = newStatus ?? oldStatus, CurrentIsStatusTransient = isTransient, PreviousStatus = oldStatus, PreviousIsStatusTransient = oldIsTransient });
            Log(LogSeverityEnum.Debug, $"{Name} - Item [{item.Id}] changed status '{oldStatus}[{oldIsTransient}]' to '{(newStatus ?? oldStatus)}[{isTransient}]'.");
        }

        protected IPsProcessProvider ResolveProvider(string providerClass, IEnumerable<ITypedArgument> arguments)
        {
            if (string.IsNullOrEmpty(providerClass))
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderRequired);

            var providerType = Type.GetType(providerClass);
            if (providerType == null)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderCannotResolve, providerClass);

            var provider = Activator.CreateInstance(providerType, arguments) as IPsProcessProvider;
            if (provider == null)
                throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderCannotResolve, providerClass);

            return provider;
        }

        protected void EnsureItemStateExpired(IPsQueueItem item)
        {
            if (item==null) 
                throw new ArgumentNullException(nameof(item));

            if (item.IsValiditySpanExpired() && item.Status != PsStatusEnum.Expired)
            {
                if (item.Status == PsStatusEnum.Started)
                {
                    if (!_workingItems.TryGetValue(item.Id, out var process))
                        throw new InvalidOperationException($"Cannot find process provider in working cache for item [{item.Id}]");
                    
                    SetItemStatusInternal(item, PsStatusEnum.Stopped, true, "Item will be stopped because expired.");
                    SetItemStatusInternal(item, PsStatusEnum.Stopped, false, "Item was stopped by system because expired.");
                    // NOTE: fast switch to stopped due prevents main loop to handle - stops later
                    _workingItems.TryRemove(item.Id, out _);
                    process.Stop();
                    SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by process.");
                }
                else
                {
                    SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by process.");
                }

                item = GetItemInternal(item.Id);
            }
        }
        protected IPsQueueItem GetItemInternal(string itemId)
        {
            ValidateItemId(itemId);

            var item = _storage.GetItem(itemId);
            return item;
        }

        protected override void OnDisposing()
        {
            base.OnDisposing();

            ProcessItemStatusChanged = null;
            _storage.Dispose();
            //TODO: Implement OnDisposing
            foreach (var item in _workingItems.Values)
                item.Stop();
            _workingItems.Clear();
            _workingItems = null!;
        }
        
        #endregion
    }
}