Merge branch 'main' of https://git.stella-ops.org/stella-ops.org/git.stella-ops.org

src/Notifier/StellaOps.Notifier/StellaOps.Notifier.Worker/Digest/DigestScheduleRunner.cs

@@ -1,252 +1,408 @@
-using System.Collections.Concurrent;
-using Cronos;
-using Microsoft.Extensions.Logging;
-using StellaOps.Notify.Models;
-using StellaOps.Notify.Storage.Mongo.Repositories;
-using StellaOps.Notifier.Worker.Channels;
-
-namespace StellaOps.Notifier.Worker.Digest;
-
-/// <summary>
-/// Default implementation of the digest schedule runner.
-/// </summary>
-public sealed class DigestScheduleRunner : IDigestScheduleRunner
-{
-    private readonly IDigestGenerator _digestGenerator;
-    private readonly INotifyChannelRepository _channelRepository;
-    private readonly IReadOnlyDictionary<NotifyChannelType, INotifyChannelAdapter> _channelAdapters;
-    private readonly TimeProvider _timeProvider;
-    private readonly ILogger<DigestScheduleRunner> _logger;
-
-    // In-memory schedule store (in production, would use a repository)
-    private readonly ConcurrentDictionary<string, DigestSchedule> _schedules = new();
-    private readonly ConcurrentDictionary<string, DateTimeOffset> _lastRunTimes = new();
-
-    public DigestScheduleRunner(
-        IDigestGenerator digestGenerator,
-        INotifyChannelRepository channelRepository,
-        IEnumerable<INotifyChannelAdapter> channelAdapters,
-        TimeProvider timeProvider,
-        ILogger<DigestScheduleRunner> logger)
-    {
-        _digestGenerator = digestGenerator ?? throw new ArgumentNullException(nameof(digestGenerator));
-        _channelRepository = channelRepository ?? throw new ArgumentNullException(nameof(channelRepository));
-        _channelAdapters = BuildAdapterMap(channelAdapters);
-        _timeProvider = timeProvider ?? TimeProvider.System;
-        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
-    }
-
-    public async Task<int> ProcessDueDigestsAsync(CancellationToken cancellationToken = default)
-    {
-        var now = _timeProvider.GetUtcNow();
-        var processed = 0;
-
-        foreach (var schedule in _schedules.Values.Where(s => s.Enabled))
-        {
-            cancellationToken.ThrowIfCancellationRequested();
-
-            try
-            {
-                if (IsDue(schedule, now))
-                {
-                    await ProcessScheduleAsync(schedule, now, cancellationToken).ConfigureAwait(false);
-                    processed++;
-                }
-            }
-            catch (Exception ex)
-            {
-                _logger.LogError(ex,
-                    "Failed to process digest schedule {ScheduleId} for tenant {TenantId}",
-                    schedule.ScheduleId, schedule.TenantId);
-            }
-        }
-
-        return processed;
-    }
-
-    public DateTimeOffset? GetNextScheduledTime(DigestSchedule schedule, DateTimeOffset? after = null)
-    {
-        ArgumentNullException.ThrowIfNull(schedule);
-
-        var referenceTime = after ?? _timeProvider.GetUtcNow();
-
-        try
-        {
-            var timeZone = TimeZoneInfo.FindSystemTimeZoneById(schedule.TimeZone);
-
-            if (!string.IsNullOrWhiteSpace(schedule.CronExpression))
-            {
-                var cron = CronExpression.Parse(schedule.CronExpression);
-                var next = cron.GetNextOccurrence(referenceTime.UtcDateTime, timeZone);
-                return next.HasValue
-                    ? new DateTimeOffset(next.Value, timeZone.GetUtcOffset(next.Value))
-                    : null;
-            }
-
-            // Default period-based scheduling
-            return schedule.Period switch
-            {
-                DigestPeriod.Hourly => referenceTime.AddHours(1).Date.AddHours(referenceTime.Hour + 1),
-                DigestPeriod.Daily => referenceTime.Date.AddDays(1).AddHours(9), // 9 AM next day
-                DigestPeriod.Weekly => GetNextWeekday(referenceTime, DayOfWeek.Monday).AddHours(9),
-                _ => null
-            };
-        }
-        catch (Exception ex)
-        {
-            _logger.LogWarning(ex,
-                "Failed to calculate next scheduled time for {ScheduleId}",
-                schedule.ScheduleId);
-            return null;
-        }
-    }
-
-    /// <summary>
-    /// Registers a digest schedule.
-    /// </summary>
-    public void RegisterSchedule(DigestSchedule schedule)
-    {
-        ArgumentNullException.ThrowIfNull(schedule);
-        _schedules[schedule.ScheduleId] = schedule;
-        _logger.LogInformation(
-            "Registered digest schedule {ScheduleId} for tenant {TenantId}",
-            schedule.ScheduleId, schedule.TenantId);
-    }
-
-    /// <summary>
-    /// Unregisters a digest schedule.
-    /// </summary>
-    public void UnregisterSchedule(string scheduleId)
-    {
-        _schedules.TryRemove(scheduleId, out _);
-        _lastRunTimes.TryRemove(scheduleId, out _);
-    }
-
-    private bool IsDue(DigestSchedule schedule, DateTimeOffset now)
-    {
-        // Check if we've run recently
-        if (_lastRunTimes.TryGetValue(schedule.ScheduleId, out var lastRun))
-        {
-            var minInterval = schedule.Period switch
-            {
-                DigestPeriod.Hourly => TimeSpan.FromMinutes(55),
-                DigestPeriod.Daily => TimeSpan.FromHours(23),
-                DigestPeriod.Weekly => TimeSpan.FromDays(6.5),
-                _ => TimeSpan.FromHours(1)
-            };
-
-            if (now - lastRun < minInterval)
-            {
-                return false;
-            }
-        }
-
-        var nextScheduled = GetNextScheduledTime(schedule, _lastRunTimes.GetValueOrDefault(schedule.ScheduleId));
-        return nextScheduled.HasValue && now >= nextScheduled.Value;
-    }
-
-    private async Task ProcessScheduleAsync(
-        DigestSchedule schedule,
-        DateTimeOffset now,
-        CancellationToken cancellationToken)
-    {
-        _logger.LogDebug("Processing digest schedule {ScheduleId}", schedule.ScheduleId);
-
-        // Calculate period
-        var (periodStart, periodEnd) = CalculatePeriod(schedule, now);
-
-        // Generate digest
-        var digest = await _digestGenerator.GenerateAsync(
-            schedule, periodStart, periodEnd, cancellationToken).ConfigureAwait(false);
-
-        // Record run time
-        _lastRunTimes[schedule.ScheduleId] = now;
-
-        // Skip if no events
-        if (digest.Status == NotifyDigestStatus.Skipped || digest.EventCount == 0)
-        {
-            _logger.LogDebug(
-                "Skipping empty digest {DigestId} for schedule {ScheduleId}",
-                digest.DigestId, schedule.ScheduleId);
-            return;
-        }
-
-        // Format content
-        var content = await _digestGenerator.FormatAsync(
-            digest, schedule.TemplateId, cancellationToken).ConfigureAwait(false);
-
-        // Get channel and send
-        var channel = await _channelRepository.GetAsync(
-            schedule.TenantId, schedule.ChannelId, cancellationToken).ConfigureAwait(false);
-
-        if (channel is null)
-        {
-            _logger.LogWarning(
-                "Channel {ChannelId} not found for digest schedule {ScheduleId}",
-                schedule.ChannelId, schedule.ScheduleId);
-            return;
-        }
-
-        if (!_channelAdapters.TryGetValue(channel.Type, out var adapter))
-        {
-            _logger.LogWarning(
-                "No adapter found for channel type {ChannelType}",
-                channel.Type);
-            return;
-        }
-
-        var rendered = NotifyDeliveryRendered.Create(
-            channelType: channel.Type,
-            format: NotifyDeliveryFormat.Json,
-            target: channel.Config?.Target ?? string.Empty,
-            title: $"Notification Digest: {schedule.Name}",
-            body: content,
-            locale: "en-us");
-
-        var result = await adapter.SendAsync(channel, rendered, cancellationToken).ConfigureAwait(false);
-
-        if (result.Success)
-        {
-            _logger.LogInformation(
-                "Sent digest {DigestId} via channel {ChannelId}: {EventCount} events",
-                digest.DigestId, schedule.ChannelId, digest.EventCount);
-        }
-        else
-        {
-            _logger.LogWarning(
-                "Failed to send digest {DigestId}: {Reason}",
-                digest.DigestId, result.Reason);
-        }
-    }
-
-    private static (DateTimeOffset Start, DateTimeOffset End) CalculatePeriod(
-        DigestSchedule schedule,
-        DateTimeOffset now)
-    {
-        return schedule.Period switch
-        {
-            DigestPeriod.Hourly => (now.AddHours(-1), now),
-            DigestPeriod.Daily => (now.Date.AddDays(-1), now.Date),
-            DigestPeriod.Weekly => (now.Date.AddDays(-7), now.Date),
-            _ => (now.AddHours(-1), now)
-        };
-    }
-
-    private static DateTimeOffset GetNextWeekday(DateTimeOffset from, DayOfWeek target)
-    {
-        var daysUntil = ((int)target - (int)from.DayOfWeek + 7) % 7;
-        if (daysUntil == 0) daysUntil = 7;
-        return from.Date.AddDays(daysUntil);
-    }
-
-    private static IReadOnlyDictionary<NotifyChannelType, INotifyChannelAdapter> BuildAdapterMap(
-        IEnumerable<INotifyChannelAdapter> adapters)
-    {
-        var builder = new Dictionary<NotifyChannelType, INotifyChannelAdapter>();
-        foreach (var adapter in adapters)
-        {
-            builder[adapter.ChannelType] = adapter;
-        }
-        return builder;
-    }
-}
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using StellaOps.Notifier.Worker.Channels;
+
+namespace StellaOps.Notifier.Worker.Digest;
+
+/// <summary>
+/// Background service that runs digest generation on configured schedules.
+/// </summary>
+public sealed class DigestScheduleRunner : BackgroundService
+{
+    private readonly IDigestGenerator _digestGenerator;
+    private readonly IDigestDistributor _distributor;
+    private readonly IDigestTenantProvider _tenantProvider;
+    private readonly DigestScheduleOptions _options;
+    private readonly TimeProvider _timeProvider;
+    private readonly ILogger<DigestScheduleRunner> _logger;
+
+    public DigestScheduleRunner(
+        IDigestGenerator digestGenerator,
+        IDigestDistributor distributor,
+        IDigestTenantProvider tenantProvider,
+        IOptions<DigestScheduleOptions> options,
+        TimeProvider timeProvider,
+        ILogger<DigestScheduleRunner> logger)
+    {
+        _digestGenerator = digestGenerator ?? throw new ArgumentNullException(nameof(digestGenerator));
+        _distributor = distributor ?? throw new ArgumentNullException(nameof(distributor));
+        _tenantProvider = tenantProvider ?? throw new ArgumentNullException(nameof(tenantProvider));
+        _options = options?.Value ?? throw new ArgumentNullException(nameof(options));
+        _timeProvider = timeProvider ?? throw new ArgumentNullException(nameof(timeProvider));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        if (!_options.Enabled)
+        {
+            _logger.LogInformation("Digest schedule runner is disabled.");
+            return;
+        }
+
+        _logger.LogInformation(
+            "Digest schedule runner started with {ScheduleCount} schedules.",
+            _options.Schedules.Count);
+
+        // Run each schedule in parallel
+        var scheduleTasks = _options.Schedules
+            .Where(s => s.Enabled)
+            .Select(schedule => RunScheduleAsync(schedule, stoppingToken))
+            .ToList();
+
+        await Task.WhenAll(scheduleTasks);
+    }
+
+    private async Task RunScheduleAsync(DigestSchedule schedule, CancellationToken stoppingToken)
+    {
+        _logger.LogInformation(
+            "Starting digest schedule '{Name}' with interval {Interval}.",
+            schedule.Name, schedule.Interval);
+
+        // Calculate initial delay to align with schedule
+        var initialDelay = CalculateInitialDelay(schedule);
+        if (initialDelay > TimeSpan.Zero)
+        {
+            _logger.LogDebug(
+                "Schedule '{Name}' waiting {Delay} for initial alignment.",
+                schedule.Name, initialDelay);
+            await Task.Delay(initialDelay, stoppingToken);
+        }
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                await ExecuteScheduleAsync(schedule, stoppingToken);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                break;
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex,
+                    "Error executing digest schedule '{Name}'. Will retry after interval.",
+                    schedule.Name);
+            }
+
+            await Task.Delay(schedule.Interval, stoppingToken);
+        }
+
+        _logger.LogInformation("Digest schedule '{Name}' stopped.", schedule.Name);
+    }
+
+    private async Task ExecuteScheduleAsync(DigestSchedule schedule, CancellationToken stoppingToken)
+    {
+        var now = _timeProvider.GetUtcNow();
+        var query = new DigestQuery
+        {
+            From = now - schedule.LookbackPeriod,
+            To = now,
+            IncludeResolved = schedule.IncludeResolved,
+            MaxIncidents = schedule.MaxIncidents,
+            ScheduleName = schedule.Name
+        };
+
+        _logger.LogDebug(
+            "Executing schedule '{Name}' for period {From} to {To}.",
+            schedule.Name, query.From, query.To);
+
+        // Get tenants to generate digests for
+        var tenants = await _tenantProvider.GetTenantsForScheduleAsync(schedule.Name, stoppingToken);
+
+        var successCount = 0;
+        var errorCount = 0;
+
+        foreach (var tenantId in tenants)
+        {
+            try
+            {
+                var result = await _digestGenerator.GenerateAsync(tenantId, query, stoppingToken);
+
+                if (!result.Summary.HasActivity && _options.SkipEmptyDigests)
+                {
+                    _logger.LogDebug(
+                        "Skipping empty digest for tenant {TenantId} on schedule '{Schedule}'.",
+                        tenantId, schedule.Name);
+                    continue;
+                }
+
+                await _distributor.DistributeAsync(result, schedule, stoppingToken);
+                successCount++;
+
+                _logger.LogInformation(
+                    "Distributed digest {DigestId} for tenant {TenantId} via schedule '{Schedule}'.",
+                    result.DigestId, tenantId, schedule.Name);
+            }
+            catch (Exception ex)
+            {
+                errorCount++;
+                _logger.LogError(ex,
+                    "Failed to generate/distribute digest for tenant {TenantId} on schedule '{Schedule}'.",
+                    tenantId, schedule.Name);
+            }
+        }
+
+        _logger.LogInformation(
+            "Completed schedule '{Name}': {Success} successful, {Errors} errors out of {Total} tenants.",
+            schedule.Name, successCount, errorCount, tenants.Count);
+    }
+
+    private TimeSpan CalculateInitialDelay(DigestSchedule schedule)
+    {
+        if (!schedule.AlignToInterval)
+        {
+            return TimeSpan.Zero;
+        }
+
+        var now = _timeProvider.GetUtcNow();
+        var intervalTicks = schedule.Interval.Ticks;
+        var currentTicks = now.Ticks;
+
+        // Calculate next aligned time
+        var nextAlignedTicks = ((currentTicks / intervalTicks) + 1) * intervalTicks;
+        var nextAligned = new DateTimeOffset(nextAlignedTicks, TimeSpan.Zero);
+
+        return nextAligned - now;
+    }
+}
+
+/// <summary>
+/// Distributes generated digests through configured channels.
+/// </summary>
+public interface IDigestDistributor
+{
+    /// <summary>
+    /// Distributes a digest via the appropriate channels.
+    /// </summary>
+    Task DistributeAsync(
+        DigestResult digest,
+        DigestSchedule schedule,
+        CancellationToken cancellationToken = default);
+}
+
+/// <summary>
+/// Provides tenants that should receive digests for a given schedule.
+/// </summary>
+public interface IDigestTenantProvider
+{
+    /// <summary>
+    /// Gets tenant IDs that should receive digests for the specified schedule.
+    /// </summary>
+    Task<IReadOnlyList<string>> GetTenantsForScheduleAsync(
+        string scheduleName,
+        CancellationToken cancellationToken = default);
+}
+
+/// <summary>
+/// Default implementation of <see cref="IDigestDistributor"/> using channel adapters.
+/// </summary>
+public sealed class ChannelDigestDistributor : IDigestDistributor
+{
+    private readonly IChannelAdapterFactory _channelFactory;
+    private readonly ILogger<ChannelDigestDistributor> _logger;
+
+    public ChannelDigestDistributor(
+        IChannelAdapterFactory channelFactory,
+        ILogger<ChannelDigestDistributor> logger)
+    {
+        _channelFactory = channelFactory ?? throw new ArgumentNullException(nameof(channelFactory));
+        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+    }
+
+    public async Task DistributeAsync(
+        DigestResult digest,
+        DigestSchedule schedule,
+        CancellationToken cancellationToken = default)
+    {
+        foreach (var channelConfig in schedule.Channels)
+        {
+            try
+            {
+                var adapter = _channelFactory.Create(channelConfig.Type);
+                var content = SelectContent(digest, channelConfig.Type);
+
+                await adapter.SendAsync(new ChannelMessage
+                {
+                    ChannelType = channelConfig.Type,
+                    Destination = channelConfig.Destination,
+                    Subject = $"Notification Digest - {digest.TenantId}",
+                    Body = content,
+                    Format = channelConfig.Format ?? GetDefaultFormat(channelConfig.Type),
+                    Metadata = new Dictionary<string, string>
+                    {
+                        ["digestId"] = digest.DigestId,
+                        ["tenantId"] = digest.TenantId,
+                        ["scheduleName"] = schedule.Name,
+                        ["from"] = digest.From.ToString("O"),
+                        ["to"] = digest.To.ToString("O")
+                    }
+                }, cancellationToken);
+
+                _logger.LogDebug(
+                    "Sent digest {DigestId} to channel {Channel} ({Destination}).",
+                    digest.DigestId, channelConfig.Type, channelConfig.Destination);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex,
+                    "Failed to send digest {DigestId} to channel {Channel}.",
+                    digest.DigestId, channelConfig.Type);
+            }
+        }
+    }
+
+    private static string SelectContent(DigestResult digest, string channelType)
+    {
+        if (digest.Content is null)
+        {
+            return string.Empty;
+        }
+
+        return channelType.ToLowerInvariant() switch
+        {
+            "slack" => digest.Content.SlackBlocks ?? digest.Content.Markdown ?? digest.Content.PlainText ?? "",
+            "email" => digest.Content.Html ?? digest.Content.PlainText ?? "",
+            "webhook" => digest.Content.Json ?? "",
+            _ => digest.Content.PlainText ?? ""
+        };
+    }
+
+    private static string GetDefaultFormat(string channelType)
+    {
+        return channelType.ToLowerInvariant() switch
+        {
+            "slack" => "blocks",
+            "email" => "html",
+            "webhook" => "json",
+            _ => "text"
+        };
+    }
+}
+
+/// <summary>
+/// In-memory implementation for testing.
+/// </summary>
+public sealed class InMemoryDigestTenantProvider : IDigestTenantProvider
+{
+    private readonly List<string> _tenants = [];
+
+    public void AddTenant(string tenantId) => _tenants.Add(tenantId);
+
+    public void RemoveTenant(string tenantId) => _tenants.Remove(tenantId);
+
+    public Task<IReadOnlyList<string>> GetTenantsForScheduleAsync(
+        string scheduleName,
+        CancellationToken cancellationToken = default)
+    {
+        return Task.FromResult<IReadOnlyList<string>>(_tenants.ToList());
+    }
+}
+
+/// <summary>
+/// Configuration options for digest scheduling.
+/// </summary>
+public sealed class DigestScheduleOptions
+{
+    /// <summary>
+    /// Configuration section name.
+    /// </summary>
+    public const string SectionName = "Notifier:DigestSchedule";
+
+    /// <summary>
+    /// Whether the digest scheduler is enabled.
+    /// </summary>
+    public bool Enabled { get; set; } = true;
+
+    /// <summary>
+    /// Whether to skip digests with no activity.
+    /// </summary>
+    public bool SkipEmptyDigests { get; set; } = true;
+
+    /// <summary>
+    /// Configured digest schedules.
+    /// </summary>
+    public List<DigestSchedule> Schedules { get; set; } = [];
+}
+
+/// <summary>
+/// A single digest schedule configuration.
+/// </summary>
+public sealed class DigestSchedule
+{
+    /// <summary>
+    /// Unique name for this schedule.
+    /// </summary>
+    public required string Name { get; set; }
+
+    /// <summary>
+    /// Whether this schedule is enabled.
+    /// </summary>
+    public bool Enabled { get; set; } = true;
+
+    /// <summary>
+    /// How often to run this schedule.
+    /// </summary>
+    public TimeSpan Interval { get; set; } = TimeSpan.FromHours(24);
+
+    /// <summary>
+    /// How far back to look for incidents.
+    /// </summary>
+    public TimeSpan LookbackPeriod { get; set; } = TimeSpan.FromHours(24);
+
+    /// <summary>
+    /// Whether to align execution to interval boundaries.
+    /// </summary>
+    public bool AlignToInterval { get; set; } = true;
+
+    /// <summary>
+    /// Maximum incidents to include.
+    /// </summary>
+    public int MaxIncidents { get; set; } = 50;
+
+    /// <summary>
+    /// Whether to include resolved incidents.
+    /// </summary>
+    public bool IncludeResolved { get; set; }
+
+    /// <summary>
+    /// Channels to distribute digests through.
+    /// </summary>
+    public List<DigestChannelConfig> Channels { get; set; } = [];
+}
+
+/// <summary>
+/// Channel configuration for digest distribution.
+/// </summary>
+public sealed class DigestChannelConfig
+{
+    /// <summary>
+    /// Channel type (email, slack, webhook, etc.).
+    /// </summary>
+    public required string Type { get; set; }
+
+    /// <summary>
+    /// Destination (email address, webhook URL, channel ID, etc.).
+    /// </summary>
+    public required string Destination { get; set; }
+
+    /// <summary>
+    /// Content format to use (html, markdown, json, blocks).
+    /// </summary>
+    public string? Format { get; set; }
+}
+
+/// <summary>
+/// Message to send through a channel.
+/// </summary>
+public sealed class ChannelMessage
+{
+    public required string ChannelType { get; init; }
+    public required string Destination { get; init; }
+    public string? Subject { get; init; }
+    public required string Body { get; init; }
+    public string Format { get; init; } = "text";
+    public IReadOnlyDictionary<string, string> Metadata { get; init; } = new Dictionary<string, string>();
+}