Finlytic/FinlyticFundamentals/Util/FundamentalsMqttClient.cs

using System;
using System.Collections.Generic;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Dtos;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Models;
using FinlyticCore.Services;
using FinlyticCore.Util;
using FinlyticFundamentals.Services;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace FinlyticFundamentals.Util;

public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
{
    private readonly ILogger<FundamentalsMqttClient> _logger;
    private readonly IConfiguration _configuration;
    private readonly IServiceScopeFactory _scopeFactory;

    public FundamentalsMqttClient(
        ILogger<FundamentalsMqttClient> logger,
        IConfiguration configuration,
        IServiceScopeFactory scopeFactory) : base(logger)
    {
        _logger = logger;
        _configuration = configuration;
        _scopeFactory = scopeFactory;
    }

    /// <inheritdoc />
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticFundamentals");

        _logger.LogInformation("[{Channel}] [MQTT_Client] Starting Fundamentals MQTT client. Host: {Host}, ClientId: {ClientId}", "MqttChannel", config.Host, config.ClientId);

        await ConnectAsync(config);
    }

    /// <inheritdoc />
    public async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("[{Channel}] [MQTT_Client] Stopping Fundamentals MQTT client.", "MqttChannel");
        await DisconnectAsync();
    }

    /// <inheritdoc />
    protected override async Task OnConnectedAsync()
    {
        _logger.LogInformation("[{Channel}] [MQTT_Client] Connected. Subscribing to RPC request topics...", "MqttChannel");
        await SubscribeAsync(MqttTopics.ResponseWildcard);
        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsGet));
        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.EventsGetAll));
        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.EventsGetByMonth));
        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsSettingsGetAll));
        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsSettingsUpdate));
        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing));

        FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
        {
            if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticFundamentals", StringComparison.OrdinalIgnoreCase))
            {
                await PublishAsync(MqttTopics.Logs("FinlyticFundamentals"), logDto);
            }
        };
    }

    /// <inheritdoc />
    protected override async Task OnMessageReceivedAsync(string topic, string payload)
    {
        if (string.IsNullOrWhiteSpace(topic)) return;

        var lastSlash = topic.LastIndexOf('/');
        if (lastSlash < 0 || lastSlash >= topic.Length - 1) return;

        var correlationId = topic.Substring(lastSlash + 1);

        if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsGet, StringComparison.OrdinalIgnoreCase))
        {
            await OnFundamentalsGetAsync(payload, correlationId);
        }
        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.EventsGetAll, StringComparison.OrdinalIgnoreCase))
        {
            await OnEventsGetAllAsync(correlationId);
        }
        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.EventsGetByMonth, StringComparison.OrdinalIgnoreCase))
        {
            await OnEventsGetByMonthAsync(payload, correlationId);
        }
        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsSettingsGetAll, StringComparison.OrdinalIgnoreCase))
        {
            await OnSettingsGetAllAsync(correlationId);
        }
        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsSettingsUpdate, StringComparison.OrdinalIgnoreCase))
        {
            await OnSettingsUpdateAsync(payload, correlationId);
        }
        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.HealthPing, StringComparison.OrdinalIgnoreCase))
        {
            await OnHealthPingAsync(topic, correlationId);
        }
    }

    private async Task OnFundamentalsGetAsync(string payload, string correlationId)
    {
        await using var scope = _scopeFactory.CreateAsyncScope();
        var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();
        var dbService = scope.ServiceProvider.GetRequiredService<IFundamentalsDbService>();

        if (string.IsNullOrWhiteSpace(payload))
        {
            await finlyticLogger.LogWarningAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Received empty payload for fundamentals_Get request.");
            return;
        }

        try
        {
            var request = (IsinRequest?)JsonSerializer.Deserialize(payload, typeof(IsinRequest), FinlyticJsonSerializerContext.Default);
            if (request == null || string.IsNullOrWhiteSpace(request.Isin))
            {
                await finlyticLogger.LogWarningAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Request missing mandatory ISIN parameter in payload.");
                return;
            }

            await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Processing RPC fundamentals_Get for ISIN '{Isin}' (forceRefresh={ForceRefresh}) [CorrelationId: {CorrelationId}]",
                request.Isin, request.ForceRefresh.ToString(), correlationId);

            var fundamentals = await dbService.GetFundamentalsAsync(request.Isin, request.Ticker, request.ForceRefresh);
            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsGet, correlationId);

            await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing RPC fundamentals response to '{ResponseTopic}'", responseTopic);
            await PublishAsync(responseTopic, fundamentals);
        }
        catch (Exception ex)
        {
            await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticFundamentals] [MQTT_Client] Failed to process fundamentals_Get request.");
        }
    }

    private async Task OnEventsGetAllAsync(string correlationId)
    {
        await using var scope = _scopeFactory.CreateAsyncScope();
        var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();
        var dbService = scope.ServiceProvider.GetRequiredService<IFundamentalsDbService>();

        await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Processing RPC events_GetAll request [CorrelationId: {CorrelationId}]", correlationId);
        try
        {
            var events = await dbService.GetAllEventsAsync();
            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.EventsGetAll, correlationId);

            await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing events RPC response to '{ResponseTopic}'", responseTopic);
            await PublishAsync(responseTopic, events);
        }
        catch (Exception ex)
        {
            await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticFundamentals] [MQTT_Client] Failed to process events_GetAll request.");
        }
    }

    private async Task OnEventsGetByMonthAsync(string payload, string correlationId)
    {
        if (string.IsNullOrWhiteSpace(payload)) return;

        await using var scope = _scopeFactory.CreateAsyncScope();
        var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();
        var dbService = scope.ServiceProvider.GetRequiredService<IFundamentalsDbService>();

        try
        {
            var request = (GetEventsByMonthRequest?)JsonSerializer.Deserialize(payload, typeof(GetEventsByMonthRequest), FinlyticJsonSerializerContext.Default);
            if (request == null) return;

            await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Processing RPC events_GetByMonth request for {Year}/{Month} [CorrelationId: {CorrelationId}]", request.Year, request.Month, correlationId);
            
            var events = await dbService.GetEventsByMonthAsync(request.Year, request.Month);
            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.EventsGetByMonth, correlationId);

            await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing monthly events RPC response to '{ResponseTopic}'", responseTopic);
            await PublishAsync(responseTopic, events);
        }
        catch (Exception ex)
        {
            await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticFundamentals] [MQTT_Client] Failed to process events_GetByMonth request.");
        }
    }

    private async Task OnSettingsGetAllAsync(string correlationId)
    {
        await using var scope = _scopeFactory.CreateAsyncScope();
        var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();

        await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
        try
        {
            var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsSettingsGetAll, correlationId);

            await PublishAsync(responseTopic, settings);
            await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
        }
        catch (Exception ex)
        {
            await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticFundamentals] [Settings_GetAll] Failed to retrieve settings.");
        }
    }

    private async Task OnSettingsUpdateAsync(string payload, string correlationId)
    {
        if (string.IsNullOrWhiteSpace(payload)) return;

        await using var scope = _scopeFactory.CreateAsyncScope();
        var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();

        await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
        try
        {
            Dictionary<string, object?>? updates = null;
            try
            {
                updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
            }
            catch
            {
                var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
                if (list != null)
                {
                    updates = new Dictionary<string, object?>();
                    foreach (var item in list)
                    {
                        updates[item.Key] = item.Value;
                    }
                }
            }

            if (updates != null && updates.Count > 0)
            {
                await settingsService.UpdateSettingsAsync(updates);
                await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
            }

            var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsSettingsUpdate, correlationId);
            await PublishAsync(responseTopic, currentSettings);
        }
        catch (Exception ex)
        {
            await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticFundamentals] [Settings_Update] Failed to update settings.");
        }
    }

    private async Task OnHealthPingAsync(string topic, string correlationId)
    {
        if (topic.Contains("FinlyticFundamentals", StringComparison.OrdinalIgnoreCase) || !topic.Contains("/", StringComparison.OrdinalIgnoreCase))
        {
            await using var scope = _scopeFactory.CreateAsyncScope();
            var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();

            var respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
            await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticFundamentals", "Online", DateTime.UtcNow, "Connected"));
            await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[FinlyticFundamentals] [Health_Ping] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
        }
    }
}