<commit_before>﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Exceptionless.Core.Extensions;
using Exceptionless.Core.Models;
using Exceptionless.Core.Repositories;
using Exceptionless.DateTimeExtensions;
using Foundatio.Caching;
using Foundatio.Jobs;
using Foundatio.Lock;
using Foundatio.Repositories;
using Foundatio.Utility;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using Microsoft.Extensions.Logging;

namespace Exceptionless.Core.Jobs {
    [Job(Description = "Closes inactive user sessions.", InitialDelay = "30s", Interval = "30s")]
    public class CloseInactiveSessionsJob : JobWithLockBase, IHealthCheck {
        private readonly IEventRepository _eventRepository;
        private readonly ICacheClient _cache;
        private readonly ILockProvider _lockProvider;
        private DateTime? _lastActivity;

        public CloseInactiveSessionsJob(IEventRepository eventRepository, ICacheClient cacheClient, ILoggerFactory loggerFactory = null) : base(loggerFactory) {
            _eventRepository = eventRepository;
            _cache = cacheClient;
            _lockProvider = new ThrottlingLockProvider(cacheClient, 1, TimeSpan.FromMinutes(1));
        }

        protected override Task<ILock> GetLockAsync(CancellationToken cancellationToken = default) {
            return _lockProvider.AcquireAsync(nameof(CloseInactiveSessionsJob), TimeSpan.FromMinutes(15), new CancellationToken(true));
        }

        protected override async Task<JobResult> RunInternalAsync(JobContext context) {
            _lastActivity = SystemClock.UtcNow;
            var results = await _eventRepository.GetOpenSessionsAsync(SystemClock.UtcNow.SubtractMinutes(1), o => o.SnapshotPaging().PageLimit(100)).AnyContext();
            int sessionsClosed = 0;
            int totalSessions = 0;
            if (results.Documents.Count == 0) {
                _logger.LogInformation("No open sessions to check.");
                return JobResult.Success;
            }

            while (results.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested) {
                var inactivePeriodUtc = SystemClock.UtcNow.Subtract(DefaultInactivePeriod);
                var sessionsToUpdate = new List<PersistentEvent>(results.Documents.Count);
                var cacheKeysToRemove = new List<string>(results.Documents.Count * 2);
                var existingSessionHeartbeatIds = new HashSet<string>(StringComparer.OrdinalIgnoreCase);

                foreach (var sessionStart in results.Documents) {
                    var lastActivityUtc = sessionStart.Date.UtcDateTime.AddSeconds((double)sessionStart.Value.GetValueOrDefault());
                    var heartbeatResult = await GetHeartbeatAsync(sessionStart).AnyContext();

                    bool closeDuplicate = heartbeatResult?.CacheKey != null && existingSessionHeartbeatIds.Contains(heartbeatResult.CacheKey);
                    if (heartbeatResult?.CacheKey != null && !closeDuplicate)
                        existingSessionHeartbeatIds.Add(heartbeatResult.CacheKey);

                    if (heartbeatResult != null && (closeDuplicate || heartbeatResult.Close || heartbeatResult.ActivityUtc > lastActivityUtc))
                        sessionStart.UpdateSessionStart(heartbeatResult.ActivityUtc, isSessionEnd: closeDuplicate || heartbeatResult.Close || heartbeatResult.ActivityUtc <= inactivePeriodUtc);
                    else if (lastActivityUtc <= inactivePeriodUtc)
                        sessionStart.UpdateSessionStart(lastActivityUtc, isSessionEnd: true);
                    else
                        continue;

                    sessionsToUpdate.Add(sessionStart);
                    if (heartbeatResult != null) {
                        cacheKeysToRemove.Add(heartbeatResult.CacheKey);
                        if (heartbeatResult.Close)
                            cacheKeysToRemove.Add(heartbeatResult.CacheKey + "-close");
                    }

                    Debug.Assert(sessionStart.Value != null && sessionStart.Value >= 0, "Session start value cannot be a negative number.");
                }

                totalSessions += results.Documents.Count;
                sessionsClosed += sessionsToUpdate.Count;

                if (sessionsToUpdate.Count > 0)
                    await _eventRepository.SaveAsync(sessionsToUpdate).AnyContext();

                if (cacheKeysToRemove.Count > 0)
                    await _cache.RemoveAllAsync(cacheKeysToRemove).AnyContext();

                _logger.LogInformation("Closing {SessionClosedCount} of {SessionCount} sessions", sessionsToUpdate, results.Documents.Count);

                // Sleep so we are not hammering the backend.<commit_msg>Fixed logging statement in sessions close job<commit_after>﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Exceptionless.Core.Extensions;
using Exceptionless.Core.Models;
using Exceptionless.Core.Repositories;
using Exceptionless.DateTimeExtensions;
using Foundatio.Caching;
using Foundatio.Jobs;
using Foundatio.Lock;
using Foundatio.Repositories;
using Foundatio.Utility;
using Microsoft.Extensions.Diagnostics.HealthChecks;
using Microsoft.Extensions.Logging;

namespace Exceptionless.Core.Jobs {
    [Job(Description = "Closes inactive user sessions.", InitialDelay = "30s", Interval = "30s")]
    public class CloseInactiveSessionsJob : JobWithLockBase, IHealthCheck {
        private readonly IEventRepository _eventRepository;
        private readonly ICacheClient _cache;
        private readonly ILockProvider _lockProvider;
        private DateTime? _lastActivity;

        public CloseInactiveSessionsJob(IEventRepository eventRepository, ICacheClient cacheClient, ILoggerFactory loggerFactory = null) : base(loggerFactory) {
            _eventRepository = eventRepository;
            _cache = cacheClient;
            _lockProvider = new ThrottlingLockProvider(cacheClient, 1, TimeSpan.FromMinutes(1));
        }

        protected override Task<ILock> GetLockAsync(CancellationToken cancellationToken = default) {
            return _lockProvider.AcquireAsync(nameof(CloseInactiveSessionsJob), TimeSpan.FromMinutes(15), new CancellationToken(true));
        }

        protected override async Task<JobResult> RunInternalAsync(JobContext context) {
            _lastActivity = SystemClock.UtcNow;
            var results = await _eventRepository.GetOpenSessionsAsync(SystemClock.UtcNow.SubtractMinutes(1), o => o.SnapshotPaging().PageLimit(100)).AnyContext();
            int sessionsClosed = 0;
            int totalSessions = 0;
            if (results.Documents.Count == 0) {
                _logger.LogInformation("No open sessions to check.");
                return JobResult.Success;
            }

            while (results.Documents.Count > 0 && !context.CancellationToken.IsCancellationRequested) {
                var inactivePeriodUtc = SystemClock.UtcNow.Subtract(DefaultInactivePeriod);
                var sessionsToUpdate = new List<PersistentEvent>(results.Documents.Count);
                var cacheKeysToRemove = new List<string>(results.Documents.Count * 2);
                var existingSessionHeartbeatIds = new HashSet<string>(StringComparer.OrdinalIgnoreCase);

                foreach (var sessionStart in results.Documents) {
                    var lastActivityUtc = sessionStart.Date.UtcDateTime.AddSeconds((double)sessionStart.Value.GetValueOrDefault());
                    var heartbeatResult = await GetHeartbeatAsync(sessionStart).AnyContext();

                    bool closeDuplicate = heartbeatResult?.CacheKey != null && existingSessionHeartbeatIds.Contains(heartbeatResult.CacheKey);
                    if (heartbeatResult?.CacheKey != null && !closeDuplicate)
                        existingSessionHeartbeatIds.Add(heartbeatResult.CacheKey);

                    if (heartbeatResult != null && (closeDuplicate || heartbeatResult.Close || heartbeatResult.ActivityUtc > lastActivityUtc))
                        sessionStart.UpdateSessionStart(heartbeatResult.ActivityUtc, isSessionEnd: closeDuplicate || heartbeatResult.Close || heartbeatResult.ActivityUtc <= inactivePeriodUtc);
                    else if (lastActivityUtc <= inactivePeriodUtc)
                        sessionStart.UpdateSessionStart(lastActivityUtc, isSessionEnd: true);
                    else
                        continue;

                    sessionsToUpdate.Add(sessionStart);
                    if (heartbeatResult != null) {
                        cacheKeysToRemove.Add(heartbeatResult.CacheKey);
                        if (heartbeatResult.Close)
                            cacheKeysToRemove.Add(heartbeatResult.CacheKey + "-close");
                    }

                    Debug.Assert(sessionStart.Value != null && sessionStart.Value >= 0, "Session start value cannot be a negative number.");
                }

                totalSessions += results.Documents.Count;
                sessionsClosed += sessionsToUpdate.Count;

                if (sessionsToUpdate.Count > 0)
                    await _eventRepository.SaveAsync(sessionsToUpdate).AnyContext();

                if (cacheKeysToRemove.Count > 0)
                    await _cache.RemoveAllAsync(cacheKeysToRemove).AnyContext();

                _logger.LogInformation("Closing {SessionClosedCount} of {SessionCount} sessions", sessionsToUpdate.Count, results.Documents.Count);

                // Sleep so we are not hammering the backend.
{'fraction_non_alphanumeric': 0.0804442599128088, 'fraction_numerical': 0.004151961801951422, 'mean_word_length': 2.754871395167576, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 25, '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': '4623488', 'n_tokens_mistral': 2390, 'n_tokens_neox': 2253, 'n_words': 479}