<commit_before>﻿using System;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using CprBroker.Engine.Tasks;
using CprBroker.Engine.Queues;
using CprBroker.Schemas;
using CprBroker.Engine;
using CprBroker.Data.Applications;

namespace CprBroker.PartInterface.Tracking
{
    public class CleanupDetectionEnqueuer : PeriodicTaskExecuter
    {
        public bool CanRunCleanup(bool log = false)
        {
            var oldestOperationTS = Operation.OldestOperationTS();
            var warmingPeriodEndTS =
                (oldestOperationTS.HasValue ? oldestOperationTS.Value : DateTime.Now)
                + SettingsUtilities.MaxInactivePeriod;

            if (DateTime.Now > warmingPeriodEndTS)
            {
                if (log)
                    Engine.Local.Admin.LogFormattedSuccess("Earliest existing operation was at <{0}>. Proceeding with cleanup", oldestOperationTS);
                return true;
            }
            else
            {
                if (log)
                    Engine.Local.Admin.LogFormattedSuccess("Cleanup cannot be performed because it is still in the warmup period. Earliest existing operation was at <{0}>. Warmup ends at <{1}>",
                        oldestOperationTS,
                        warmingPeriodEndTS
                    );
                return false;
            }
        }

        protected override void PerformTimerAction()
        {
            // Register an operation
            BrokerContext.Current.RegisterOperation(CprBroker.Data.Applications.OperationType.Types.Generic, this.GetType().Name);

            // Cold-start protection 
            // Only perform cleanup if the system has been running for a minimum of the data storage limit (a.k.a 3 months)
            if (CanRunCleanup(log: true) == false)
            {
                return;
            }

            var startIndex = 0;
            var foundUuids = new PersonIdentifier[0];
            var maximumUsageDate = DateTime.Now;
            var minimumUsageDate = maximumUsageDate - SettingsUtilities.MaxInactivePeriod;
            var cleanupQueue = Queue.GetQueues<CleanupQueue>().First();

            do
            {
                var prov = new TrackingDataProvider();
                if (LogTimerEvents)
                    CprBroker.Engine.Local.Admin.LogFormattedSuccess("{0}: checking persons {1} to {2}", GetType().Name, startIndex, startIndex + BatchSize);
                foundUuids = prov.EnumeratePersons(startIndex, BatchSize);

                var queueItems = foundUuids
                    .Select(uuid => new CleanupQueueItem() { PersonUuid = uuid.UUID.Value, PNR = uuid.CprNumber })
                    .ToArray();

                if (LogTimerEvents || foundUuids.Length > 0)
                    CprBroker.Engine.Local.Admin.LogFormattedSuccess("{0}: {1} persons enqueued for inspection <{2}>",
                        GetType().Name,
                        foundUuids.Length,
                        string.Join(",", foundUuids.Select(id => id.UUID))
                        );

                cleanupQueue.Enqueue(queueItems);
                startIndex += BatchSize;

            } while (foundUuids.Length > 0 && foundUuids.Length < BatchSize);
        }

    }

}<commit_msg>Bugfix: synchronisation of Cleanup detection * Fix for loop continuation * Queue items are guarded by a semaphore that is only signalled after all persons have been enqueued<commit_after>﻿using System;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using CprBroker.Engine.Tasks;
using CprBroker.Engine.Queues;
using CprBroker.Schemas;
using CprBroker.Engine;
using CprBroker.Data.Applications;

namespace CprBroker.PartInterface.Tracking
{
    public class CleanupDetectionEnqueuer : PeriodicTaskExecuter
    {
        public bool CanRunCleanup(bool log = false)
        {
            var oldestOperationTS = Operation.OldestOperationTS();
            var warmingPeriodEndTS =
                (oldestOperationTS.HasValue ? oldestOperationTS.Value : DateTime.Now)
                + SettingsUtilities.MaxInactivePeriod;

            if (DateTime.Now > warmingPeriodEndTS)
            {
                if (log)
                    Engine.Local.Admin.LogFormattedSuccess("Earliest existing operation was at <{0}>. Proceeding with cleanup", oldestOperationTS);
                return true;
            }
            else
            {
                if (log)
                    Engine.Local.Admin.LogFormattedSuccess("Cleanup cannot be performed because it is still in the warmup period. Earliest existing operation was at <{0}>. Warmup ends at <{1}>",
                        oldestOperationTS,
                        warmingPeriodEndTS
                    );
                return false;
            }
        }

        protected override void PerformTimerAction()
        {
            // Register an operation
            BrokerContext.Current.RegisterOperation(CprBroker.Data.Applications.OperationType.Types.Generic, this.GetType().Name);

            // Cold-start protection 
            // Only perform cleanup if the system has been running for a minimum of the data storage limit (a.k.a 3 months)
            if (CanRunCleanup(log: true) == false)
            {
                return;
            }

            var startIndex = 0;
            var foundUuids = new PersonIdentifier[0];
            var maximumUsageDate = DateTime.Now;
            var minimumUsageDate = maximumUsageDate - SettingsUtilities.MaxInactivePeriod;
            var cleanupQueue = Queue.GetQueues<CleanupQueue>().First();

            // Create a semaphore to block queue processing until all person UUID's have been enumerated
            var thisSemaphore = Semaphore.Create();

            do
            {
                var prov = new TrackingDataProvider();
                if (LogTimerEvents)
                    CprBroker.Engine.Local.Admin.LogFormattedSuccess("{0}: checking persons {1} to {2}", GetType().Name, startIndex, startIndex + BatchSize);
                foundUuids = prov.EnumeratePersons(startIndex, BatchSize);

                var queueItems = foundUuids
                    .Select(uuid => new CleanupQueueItem() { PersonUuid = uuid.UUID.Value, PNR = uuid.CprNumber })
                    .ToArray();

                if (LogTimerEvents || foundUuids.Length > 0)
                    CprBroker.Engine.Local.Admin.LogFormattedSuccess("{0}: {1} persons enqueued for inspection <{2}>",
                        GetType().Name,
                        foundUuids.Length,
                        string.Join(",", foundUuids.Select(id => id.UUID))
                        );

                cleanupQueue.Enqueue(queueItems, thisSemaphore);
                startIndex += BatchSize;

            } while (foundUuids.Length == BatchSize);

            // Now signal the semaphore
            thisSemaphore.Signal();
        }

    }

}
{'fraction_non_alphanumeric': 0.08321315258148256, 'fraction_numerical': 0.003893856359965388, 'mean_word_length': 1.8225478225478224, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5009402', 'n_tokens_mistral': 1819, 'n_tokens_neox': 1733, 'n_words': 432}