up

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

@@ -0,0 +1,408 @@
+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>();
+}