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 from FinlyticNews.
/// </summary>
public class SentimentBackgroundService : BackgroundService
{
    private readonly SentimentMqttClient _mqttClient;
    private readonly IFinBertAnalyzerService _analyzer;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IFinlyticLogger<SentimentBackgroundService> _finlyticLogger;

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

    public SentimentBackgroundService(
        SentimentMqttClient mqttClient,
        IFinBertAnalyzerService analyzer,
        IServiceScopeFactory scopeFactory,
        IFinlyticLogger<SentimentBackgroundService> finlyticLogger)
    {
        _mqttClient = mqttClient;
        _analyzer = analyzer;
        _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;

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

            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);

            try
            {
                await Task.Delay(interval, stoppingToken);
            }
            catch (OperationCanceledException)
            {
                break;
            }
        }

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

    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;

            using var scope = _scopeFactory.CreateScope();
            var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();

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

                    await dbService.SaveArticleSentimentAsync(
                        articleId: article.Id,
                        isin: asset.Isin,
                        companyName: asset.Name,
                        sector: null,
                        publishedAt: article.PublishedAt,
                        finbert: finbert,
                        ct: cancellationToken
                    );

                    // Broadcast real-time updated summary for this asset
                    var summaryDto = await dbService.GetIsinSummaryDtoAsync(asset.Isin, cancellationToken);
                    if (summaryDto != null)
                    {
                        await _mqttClient.BroadcastSentimentResultAsync(asset.Isin, summaryDto);
                    }
                }
            }

            await _mqttClient.UpdateArticleStatusAsync(article.Id, "Analyzed");
            await _finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentBackgroundService] Completed sentiment persistence and status transition for article {Id}.", article.Id);
        }
        catch (Exception ex)
        {
            await _finlyticLogger.LogErrorAsync(SettingKeys.SentimentChannel, ex, "[SentimentBackgroundService] Failed to process sentiment for article: {Id}", article.Id);
        }
        finally
        {
            ProcessingArticles.TryRemove(article.Id, out _);
        }
    }
}