EonaCat.LogStack/EonaCat.LogStack/EonaCatLoggerCore/Flows/AggregationFlow.cs

using EonaCat.LogStack.Core;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace EonaCat.LogStack.Flows
{
    // This file is part of the EonaCat project(s) which is released under the Apache License.
    // See the LICENSE file or go to https://EonaCat.com/License for full license details.

    /// <summary>
    /// A flow that aggregates log events and periodically emits summary logs.
    /// Groups logs by category or level and provides statistics.
    /// </summary>
    public sealed class AggregationFlow : FlowBase
    {
        private readonly IFlow _innerFlow;
        private readonly TimeSpan _aggregationInterval;
        private readonly Dictionary<string, AggregationStats> _stats;
        private readonly Timer _flushTimer;
        private readonly object _statsLock = new();
        private bool _disposed;

        public AggregationFlow(
            IFlow innerFlow,
            int aggregationIntervalMs = 60000,
            LogLevel minimumLevel = LogLevel.Trace)
            : base(innerFlow?.Name + "_Aggregated" ?? "AggregationFlow", minimumLevel)
        {
            _innerFlow = innerFlow ?? throw new ArgumentNullException(nameof(innerFlow));
            _aggregationInterval = TimeSpan.FromMilliseconds(aggregationIntervalMs > 0 ? aggregationIntervalMs : 60000);
            _stats = new Dictionary<string, AggregationStats>();
            _flushTimer = new Timer(FlushAggregationCallback, null, _aggregationInterval, _aggregationInterval);
        }

        private class AggregationStats
        {
            public int Count { get; set; }
            public int ErrorCount { get; set; }
            public int WarningCount { get; set; }
            public DateTime FirstOccurrence { get; set; }
            public DateTime LastOccurrence { get; set; }
            public HashSet<string> Messages { get; } = new HashSet<string>();
        }

        public override async Task<WriteResult> BlastAsync(LogEvent logEvent, CancellationToken cancellationToken = default)
        {
            if (!IsEnabled || !IsLogLevelEnabled(logEvent))
            {
                return WriteResult.LevelFiltered;
            }

            lock (_statsLock)
            {
                var key = $"{logEvent.Category}_{logEvent.Level}";

                if (!_stats.TryGetValue(key, out var stats))
                {
                    stats = new AggregationStats { FirstOccurrence = DateTime.UtcNow };
                    _stats[key] = stats;
                }

                stats.Count++;
                stats.LastOccurrence = DateTime.UtcNow;

                if (logEvent.Level >= LogLevel.Error)
                {
                    stats.ErrorCount++;
                }
                else if (logEvent.Level == LogLevel.Warning)
                {
                    stats.WarningCount++;
                }

                var messageStr = logEvent.Message.ToString();
                if (!string.IsNullOrEmpty(messageStr) && stats.Messages.Count < 10)
                {
                    stats.Messages.Add(messageStr);
                }
            }

            Interlocked.Increment(ref BlastedCount);
            return WriteResult.Success;
        }

        public override async Task<WriteResult> BlastBatchAsync(ReadOnlyMemory<LogEvent> logEvents, CancellationToken cancellationToken = default)
        {
            if (!IsEnabled)
            {
                return WriteResult.FlowDisabled;
            }

            var eventsArray = logEvents.ToArray();
            foreach (var logEvent in eventsArray)
            {
                if (IsLogLevelEnabled(logEvent))
                {
                    await BlastAsync(logEvent, cancellationToken).ConfigureAwait(false);
                }
            }

            return WriteResult.Success;
        }

        public override async Task FlushAsync(CancellationToken cancellationToken = default)
        {
            await EmitAggregationSummaryAsync(cancellationToken).ConfigureAwait(false);
        }

        private async void FlushAggregationCallback(object? state)
        {
            if (!_disposed)
            {
                await EmitAggregationSummaryAsync(CancellationToken.None).ConfigureAwait(false);
            }
        }

        private async Task EmitAggregationSummaryAsync(CancellationToken cancellationToken)
        {
            Dictionary<string, AggregationStats> currentStats;
            lock (_statsLock)
            {
                if (_stats.Count == 0)
                {
                    return;
                }

                currentStats = new Dictionary<string, AggregationStats>(_stats);
                _stats.Clear();
            }

            var summaryMessages = new List<string>();
            summaryMessages.Add($"=== Log Aggregation Summary (Period: {_aggregationInterval.TotalSeconds:F0}s) ===");

            foreach (var kvp in currentStats.OrderBy(x => x.Key))
            {
                var key = kvp.Key;
                var stats = kvp.Value;
                var duration = (stats.LastOccurrence - stats.FirstOccurrence).TotalSeconds;

                summaryMessages.Add(
                    $"Category-Level: {key} | Count: {stats.Count} | " +
                    $"Errors: {stats.ErrorCount} | Warnings: {stats.WarningCount} | " +
                    $"Duration: {duration:F2}s");

                if (stats.Messages.Count > 0)
                {
                    summaryMessages.Add($"  Sample messages: {string.Join("; ", stats.Messages.Take(3))}");
                }
            }

            var summaryMessage = string.Join(Environment.NewLine, summaryMessages);

            var summaryEvent = new LogEvent
            {
                Message = summaryMessage.AsMemory(),
                Category = "AggregationFlow",
                Level = LogLevel.Information,
                Timestamp = LogEvent.CreateTimestamp(DateTime.UtcNow),
                Exception = null,
                Properties = new Dictionary<string, object>
                {
                    { "AggregationType", "Summary" },
                    { "StatsCount", currentStats.Count }
                }
            };

            try
            {
                await _innerFlow.BlastAsync(summaryEvent, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                System.Diagnostics.Debug.WriteLine($"AggregationFlow summary emit error: {ex.Message}");
            }
        }

        public override async ValueTask DisposeAsync()
        {
            _disposed = true;
            _flushTimer?.Dispose();
            await EmitAggregationSummaryAsync(default).ConfigureAwait(false);
            await _innerFlow.DisposeAsync().ConfigureAwait(false);
            await base.DisposeAsync().ConfigureAwait(false);
        }
    }
}