Finlytic/FinlyticSentiment/Services/SentimentBackgroundService.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Dtos.News;
using FinlyticCore.Services;
using FinlyticSentiment.Util;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

namespace FinlyticSentiment.Services;

/// <summary>
/// Background hosted worker executing periodic sentiment analysis sweeps on pending news articles
/// and processing real-time article broadcasts.
/// </summary>
public class SentimentBackgroundService : BackgroundService
{
    private readonly SentimentMqttClient _mqttClient;
    private readonly IFinBertAnalyzerService _analyzer;
    private readonly ISentimentStorageService _storage;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IFinlyticLogger<SentimentBackgroundService> _finlyticLogger;

    private static readonly ConcurrentDictionary<Guid, byte> ProcessingArticles = new();

    public SentimentBackgroundService(
        SentimentMqttClient mqttClient,
        IFinBertAnalyzerService analyzer,
        ISentimentStorageService storage,
        IServiceScopeFactory scopeFactory,
        IFinlyticLogger<SentimentBackgroundService> finlyticLogger)
    {
        _mqttClient = mqttClient;
        _analyzer = analyzer;
        _storage = storage;
        _scopeFactory = scopeFactory;
        _finlyticLogger = finlyticLogger;
    }

    /// <inheritdoc />
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinlyticSentiment Background Service started.");

        _mqttClient.OnArticleReceived += async (article) =>
        {
            await ProcessSingleArticleAsync(article, stoppingToken);
        };

        await Task.Delay(3000, stoppingToken);

        while (!stoppingToken.IsCancellationRequested)
        {
            int maxBatchSize = 10;
            int sweepIntervalMinutes = 5;

            using (var scope = _scopeFactory.CreateScope())
            {
                var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
                maxBatchSize = await settings.GetSettingAsync(SettingKeys.MaxBatchSize, stoppingToken);
            }

            var interval = TimeSpan.FromMinutes(Math.Max(1, sweepIntervalMinutes));

            try
            {
                await PerformSentimentSweepAsync(maxBatchSize, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Unhandled exception encountered during sentiment sweep cycle.");
            }

            await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Waiting {Minutes} minute(s) until next sentiment sweep...", interval.TotalMinutes);

            using var timer = new PeriodicTimer(interval);
            try
            {
                await timer.WaitForNextTickAsync(stoppingToken);
            }
            catch (OperationCanceledException)
            {
                break;
            }
        }

        await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinlyticSentiment Background Service is shutting down gracefully.");
    }

    private async Task PerformSentimentSweepAsync(int maxBatchSize, CancellationToken cancellationToken)
    {
        await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Starting sentiment sweep for pending news articles (Limit: {Limit})...", maxBatchSize);

        List<NewsArticleDto> pendingArticles = await _mqttClient.GetPendingArticlesAsync(limit: maxBatchSize);
        if (pendingArticles.Count == 0)
        {
            await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] No pending news articles found in FinlyticNews.");
            return;
        }

        await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Retrieved {Count} pending article(s) for FinBERT sentiment evaluation.", pendingArticles.Count);

        foreach (var article in pendingArticles)
        {
            if (cancellationToken.IsCancellationRequested) break;
            await ProcessSingleArticleAsync(article, cancellationToken);
        }
    }

    private async Task ProcessSingleArticleAsync(NewsArticleDto article, CancellationToken cancellationToken = default)
    {
        if (article == null || article.Id == Guid.Empty) return;

        if (!ProcessingArticles.TryAdd(article.Id, 0))
        {
            await _finlyticLogger.LogDebugAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Article {Id} is already being processed. Skipping duplicate run.", article.Id);
            return;
        }

        try
        {
            await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Evaluating FinBERT sentiment for article: '{Title}' (ID: {Id})", article.Title, article.Id);

            var finbert = await _analyzer.AnalyzeArticleAsync(article);

            if (finbert == null)
            {
                await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] FinBERT analysis returned NULL for article {Id} ('{Title}'). Aborting processing for this run.", article.Id, article.Title);
                return;
            }

            if (cancellationToken.IsCancellationRequested) return;

            await _storage.SaveArticleSentimentAsync(article, finbert);

            if (article.MatchedAssets != null && article.MatchedAssets.Count > 0)
            {
                foreach (var asset in article.MatchedAssets)
                {
                    if (cancellationToken.IsCancellationRequested) break;
                    if (string.IsNullOrWhiteSpace(asset.Isin)) continue;

                    await _storage.UpdateIsinSummaryAsync(
                        asset.Isin,
                        asset.Name,
                        "General",
                        article,
                        finbert);

                    await _storage.UpdateSectorSummaryAsync(
                        "General",
                        asset.Isin,
                        article.Id.ToString(),
                        finbert);
                }
            }

            if (cancellationToken.IsCancellationRequested) return;

            bool updated = await _mqttClient.UpdateArticleStatusAsync(article.Id, "Analyzed");
            if (updated)
            {
                await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Article sentiment processed and status set to 'Analyzed' in FinlyticNews: {Title} (ID: {Id}) -> {Label}", article.Title, article.Id, finbert.Label);
            }
            else
            {
                await _finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Failed to confirm status update to 'Analyzed' in FinlyticNews for article: {Id}", article.Id);
            }
        }
        catch (Exception ex)
        {
            await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Error processing sentiment for article: {Id} ({Title})", article.Id, article.Title);
        }
        finally
        {
            ProcessingArticles.TryRemove(article.Id, out _);
        }
    }
}