272 lines
12 KiB
C#
272 lines
12 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FinlyticCore.Dtos;
|
|
using FinlyticCore.Dtos.News;
|
|
using FinlyticCore.Dtos.Sentiment;
|
|
using FinlyticCore.Dtos.Settings;
|
|
using FinlyticCore.Models;
|
|
using FinlyticCore.Services;
|
|
using FinlyticCore.Util;
|
|
using FinlyticNews.Entities;
|
|
using FinlyticNews.Services;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace FinlyticNews.Util;
|
|
|
|
/// <summary>
|
|
/// Managed MQTT client for broadcasting completed news articles and handling typed RPC requests for FinlyticNews.
|
|
/// </summary>
|
|
public class NewsMqttClient : ManagedMqttClient, IHostedService
|
|
{
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly ILogger<NewsMqttClient> _logger;
|
|
private readonly IConfiguration _configuration;
|
|
|
|
public NewsMqttClient(
|
|
ILogger<NewsMqttClient> logger,
|
|
IServiceScopeFactory scopeFactory,
|
|
IConfiguration configuration) : base(logger)
|
|
{
|
|
_scopeFactory = scopeFactory;
|
|
_logger = logger;
|
|
_configuration = configuration;
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public async Task StartAsync(CancellationToken cancellationToken)
|
|
{
|
|
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticNews");
|
|
|
|
_logger.LogInformation("Starting News MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
|
await ConnectAsync(config);
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.LogInformation("Stopping News MQTT client and disconnecting.");
|
|
await DisconnectAsync();
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override async Task OnConnectedAsync()
|
|
{
|
|
_logger.LogInformation("News MQTT client connected. Registering RPC topic subscriptions...");
|
|
|
|
await SubscribeAsync(MqttTopics.ResponseWildcard);
|
|
await SubscribeRpcAsync<DailyNewsRequest, List<NewsArticleDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsGet), HandleNewsGetRpcAsync);
|
|
await SubscribeRpcAsync<ArticleRequest, NewsArticleDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsGetById), HandleNewsGetByIdRpcAsync);
|
|
await SubscribeRpcAsync<object, List<NewsArticleDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsGetPending), HandleNewsGetPendingRpcAsync);
|
|
await SubscribeRpcAsync<UpdateNewsStatusRequest, UpdateNewsStatusResponse>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsUpdateStatus), HandleNewsUpdateStatusRpcAsync);
|
|
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsSettingsGetAll), HandleSettingsGetAllRpcAsync);
|
|
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsSettingsUpdate), HandleSettingsUpdateRpcAsync);
|
|
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
|
|
|
|
FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
|
{
|
|
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticNews", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await PublishAsync(MqttTopics.Logs("FinlyticNews"), logDto);
|
|
}
|
|
};
|
|
}
|
|
|
|
/// <summary>
|
|
/// Broadcasts a newly processed news article to downstream subscribers.
|
|
/// </summary>
|
|
public async Task BroadcastArticleAsync(NewsArticleDto article)
|
|
{
|
|
const string topic = MqttTopics.NewsCompleted;
|
|
_logger.LogInformation("Broadcasting completed article to MQTT topic: {Topic}. ID: {Id}", topic, article.Id);
|
|
await PublishAsync(topic, article);
|
|
|
|
var firstIsin = article.MatchedAssets.FirstOrDefault()?.Isin;
|
|
if (!string.IsNullOrWhiteSpace(firstIsin))
|
|
{
|
|
string isinTopic = MqttTopics.NewsStream(firstIsin);
|
|
await PublishAsync(isinTopic, article);
|
|
}
|
|
}
|
|
|
|
private async Task<List<NewsArticleDto>> HandleNewsGetRpcAsync(DailyNewsRequest? req, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "Received RPC news_Get request [CorrelationId: {CorrelationId}]", correlationId);
|
|
|
|
int limit = req?.Limit > 0 ? req.Limit : 20;
|
|
int offset = req?.Offset >= 0 ? req.Offset : 0;
|
|
string? isin = !string.IsNullOrWhiteSpace(req?.Isin) ? req.Isin : null;
|
|
string? status = req?.Status;
|
|
string? searchQuery = req?.Query;
|
|
DateTime? date = req?.Date;
|
|
|
|
if (req?.HasSentiment == true && string.IsNullOrEmpty(status))
|
|
{
|
|
status = "Analyzed";
|
|
}
|
|
|
|
try
|
|
{
|
|
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
|
var articles = await dbService.GetFilteredNewsAsync(limit, offset, isin, date, status, searchQuery);
|
|
return (await Task.WhenAll(articles.Select(a => MapToDtoAsync(a)))).ToList();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "Failed to compile RPC response for news_Get");
|
|
return [];
|
|
}
|
|
}
|
|
|
|
private async Task<NewsArticleDto?> HandleNewsGetByIdRpcAsync(ArticleRequest? req, string correlationId)
|
|
{
|
|
var targetIdStr = req?.ArticleId ?? req?.Id;
|
|
if (Guid.TryParse(targetIdStr, out var articleId))
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
|
var article = await dbService.GetArticleByIdAsync(articleId);
|
|
|
|
if (article != null)
|
|
{
|
|
return await MapToDtoAsync(article);
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
private async Task<List<NewsArticleDto>> HandleNewsGetPendingRpcAsync(object? _, string correlationId)
|
|
{
|
|
try
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
|
var pendingArticles = await dbService.GetArticlesByStatusAsync("Completed");
|
|
return (await Task.WhenAll(pendingArticles.Take(10).Select(a => MapToDtoAsync(a)))).ToList();
|
|
}
|
|
catch
|
|
{
|
|
return [];
|
|
}
|
|
}
|
|
|
|
private async Task<UpdateNewsStatusResponse> HandleNewsUpdateStatusRpcAsync(UpdateNewsStatusRequest? req, string correlationId)
|
|
{
|
|
if (req != null && req.Id != Guid.Empty)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
|
await dbService.UpdateArticleStatusAsync(req.Id, req.Status);
|
|
return new UpdateNewsStatusResponse(true, "Status updated successfully.");
|
|
}
|
|
return new UpdateNewsStatusResponse(false, "Invalid payload.");
|
|
}
|
|
|
|
private async Task<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_GetAll] Retrieving service dynamic settings [CorrelationId: {CorrelationId}]", correlationId);
|
|
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
|
}
|
|
|
|
private async Task<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
|
|
if (updates != null && updates.Count > 0)
|
|
{
|
|
await settingsService.UpdateSettingsAsync(updates);
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
|
}
|
|
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
|
}
|
|
|
|
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
|
|
{
|
|
if (topic.Contains("FinlyticNews", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
|
|
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticNews", "Online", DateTime.UtcNow, "Connected"));
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Maps a NewsArticleEntity to a NewsArticleDto, enriching it with the FinBERT sentiment result for the
|
|
/// article by querying FinlyticSentiment over the <see cref="MqttTopics.Channels.SentimentGetArticle"/> RPC
|
|
/// channel. This previously read a legacy on-disk cache at <c>data/summaries/articles/*.json</c> and
|
|
/// <c>data/summaries/isin/*.json</c>, but nothing in this repository writes those files anymore (FinlyticSentiment
|
|
/// persists results to its own database) - the disk lookup was silently degrading Sentiment to always-null.
|
|
/// </summary>
|
|
private async Task<NewsArticleDto> MapToDtoAsync(NewsArticleEntity a)
|
|
{
|
|
string? sentimentLabel = null;
|
|
double? sentimentScore = null;
|
|
double? confidence = null;
|
|
FinBertResultDto? finbertResult = null;
|
|
|
|
try
|
|
{
|
|
if (IsConnected)
|
|
{
|
|
var targetId = a.Id.ToString();
|
|
var sentimentEntry = await SendRpcRequestAsync<IsinAnalysisEntry, ArticleRequest>(
|
|
MqttTopics.Channels.SentimentGetArticle,
|
|
new ArticleRequest(targetId, targetId),
|
|
TimeSpan.FromSeconds(3));
|
|
|
|
if (sentimentEntry?.FinbertResult != null)
|
|
{
|
|
finbertResult = sentimentEntry.FinbertResult;
|
|
sentimentLabel = finbertResult.Label;
|
|
sentimentScore = finbertResult.CompoundScore;
|
|
confidence = finbertResult.Confidence;
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "[NewsMqttClient] Failed to fetch sentiment for article {ArticleId} via '{Channel}' RPC.", a.Id, MqttTopics.Channels.SentimentGetArticle);
|
|
}
|
|
|
|
return new NewsArticleDto
|
|
{
|
|
Id = a.Id,
|
|
Title = a.Title,
|
|
Author = a.Author,
|
|
Summary = a.Summary,
|
|
ContentRaw = a.ContentRaw,
|
|
Language = a.Language,
|
|
SourceUrl = a.SourceUrl,
|
|
ScrapedAt = a.ScrapedAt,
|
|
PublishedAt = a.PublishedAt,
|
|
Status = a.Status,
|
|
Sentiment = sentimentLabel,
|
|
SentimentScore = sentimentScore,
|
|
Confidence = confidence,
|
|
FinbertResult = finbertResult,
|
|
MatchedAssets = (a.MatchedAssets ?? []).Select(m => new MatchedAssetDto
|
|
{
|
|
Name = m.Name,
|
|
Isin = m.Isin
|
|
}).ToList()
|
|
};
|
|
}
|
|
} |