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.Models;
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 IFundamentalsDbService _dbService;
    private readonly IServiceScopeFactory _scopeFactory;

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

    /// <inheritdoc />
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var config = new MqttConfiguration
        {
            Host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost",
            Port = Convert.ToInt32(_configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883"),
            ClientId = _configuration["MQTT:ClientId"] ?? "finlytic_fundamentals_" + Guid.NewGuid().ToString("N")
        };

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

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

    /// <inheritdoc />
    protected override async Task OnConnectedAsync()
    {
        _logger.LogInformation("[{Channel}] Fundamentals MQTT client connected. Subscribing to RPC request topics...", "FundamentalsChannel");
        await SubscribeAsync("services/request/fundamentals_Get/#");
        await SubscribeAsync("services/request/events_GetAll/#");
        await SubscribeAsync("services/request/health_Ping/#");
        await SubscribeAsync("services/config/updated/#");
    }

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

        // 1. Config update events
        if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
        {
            if (topic.EndsWith("FinlyticFundamentals", StringComparison.OrdinalIgnoreCase))
            {
                await OnConfigUpdatedAsync(payload);
            }
            return;
        }

        // Extract correlationId from topic suffix (e.g. services/request/fundamentals_Get/{correlationId})
        var lastSlash = topic.LastIndexOf('/');
        if (lastSlash < 0 || lastSlash >= topic.Length - 1) return;

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

        // 2. Dispatch to specific channel handlers
        if (topic.Contains("fundamentals_Get", StringComparison.OrdinalIgnoreCase))
        {
            await OnFundamentalsGetAsync(payload, correlationId);
        }
        else if (topic.Contains("events_GetAll", StringComparison.OrdinalIgnoreCase))
        {
            await OnEventsGetAllAsync(correlationId);
        }
        else if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
        {
            await OnHealthPingAsync(topic, correlationId);
        }
    }

    /// <summary>
    /// Handles fundamentals_Get RPC requests using source-generated DTO deserialization.
    /// </summary>
    private async Task OnFundamentalsGetAsync(string payload, string correlationId)
    {
        if (string.IsNullOrWhiteSpace(payload))
        {
            _logger.LogWarning("[{Channel}] [FundamentalsMqttClient] Received empty payload for fundamentals_Get request.", "FundamentalsChannel");
            return;
        }

        try
        {
            var request = (IsinRequest?)JsonSerializer.Deserialize(payload, typeof(IsinRequest), FinlyticJsonSerializerContext.Default);
            if (request == null || string.IsNullOrWhiteSpace(request.Isin))
            {
                _logger.LogWarning("[{Channel}] [FundamentalsMqttClient] Request missing mandatory ISIN parameter in payload.", "FundamentalsChannel");
                return;
            }

            _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Processing RPC fundamentals_Get for ISIN '{Isin}' (forceRefresh={ForceRefresh}) [CorrelationId: {CorrelationId}]",
                "FundamentalsChannel", request.Isin, request.ForceRefresh.ToString(), correlationId);

            var fundamentals = await _dbService.GetFundamentalsAsync(request.Isin, request.Ticker, request.ForceRefresh);
            var responseTopic = $"services/response/fundamentals_Get/{correlationId}";

            _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Publishing RPC fundamentals response to '{ResponseTopic}'", "FundamentalsChannel", responseTopic);
            await PublishAsync(responseTopic, fundamentals);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[{Channel}] [FundamentalsMqttClient] Failed to process fundamentals_Get request.", "FundamentalsChannel");
        }
    }

    /// <summary>
    /// Handles events_GetAll RPC requests.
    /// </summary>
    private async Task OnEventsGetAllAsync(string correlationId)
    {
        _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Processing RPC events_GetAll request [CorrelationId: {CorrelationId}]", "FundamentalsChannel", correlationId);
        try
        {
            var events = await _dbService.GetAllEventsAsync();
            var responseTopic = $"services/response/events_GetAll/{correlationId}";

            _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Publishing events RPC response to '{ResponseTopic}'", "FundamentalsChannel", responseTopic);
            await PublishAsync(responseTopic, events);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[{Channel}] [FundamentalsMqttClient] Failed to process events_GetAll request.", "FundamentalsChannel");
        }
    }

    /// <summary>
    /// Handles health_Ping RPC requests.
    /// </summary>
    private async Task OnHealthPingAsync(string topic, string correlationId)
    {
        if (topic.Contains("FinlyticFundamentals", StringComparison.OrdinalIgnoreCase) || !topic.Contains("/", StringComparison.OrdinalIgnoreCase))
        {
            string respTopic = $"services/response/health_Ping/{correlationId}";
            await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticFundamentals", "Online", DateTime.UtcNow, "Connected"));
            _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "FundamentalsChannel", correlationId);
        }
    }

    /// <summary>
    /// Handles dynamic service config update events.
    /// </summary>
    private async Task OnConfigUpdatedAsync(string payload)
    {
        _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Received config update event for FinlyticFundamentals.", "FundamentalsChannel");
        try
        {
            using var doc = JsonDocument.Parse(payload);
            if (doc.RootElement.TryGetProperty("settings", out var settingsProp))
            {
                var dict = (Dictionary<string, string>?)JsonSerializer.Deserialize(settingsProp.GetRawText(), typeof(Dictionary<string, string>), FinlyticJsonSerializerContext.Default);
                if (dict != null && dict.Count > 0)
                {
                    using var scope = _scopeFactory.CreateScope();
                    var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
                    await settingsDb.UpdateSettingsFromDictionaryAsync(dict);
                    _logger.LogInformation("[{Channel}] [FundamentalsMqttClient] Persisted {Count} updated settings to FinlyticFundamentals database.", "FundamentalsChannel", dict.Count);
                }
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[{Channel}] [FundamentalsMqttClient] Error processing MQTT config update event.", "FundamentalsChannel");
        }
    }
}