Finlytic/FinlyticTechnicalAnalysis/Util/TAMqttClient.cs

using System;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Models;
using FinlyticCore.Util;
using FinlyticTechnicalAnalysis.Services;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace FinlyticTechnicalAnalysis.Util;

public class TAMqttClient(
    ILogger<TAMqttClient> logger,
    IConfiguration configuration,
    IServiceScopeFactory scopeFactory) : ManagedMqttClient(logger), IHostedService
{
    /// <summary>
    /// Starts the MQTT client.
    /// </summary>
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var host = configuration["MQTT:Host"] ?? configuration["MQTT__Host"] ?? "localhost";
        var portStr = configuration["MQTT:Port"] ?? configuration["MQTT__Port"] ?? "1883";
        var clientId = configuration["MQTT:ClientId"] ?? "finlytic_ta_" + Guid.NewGuid().ToString("N");

        var config = new MqttConfiguration
        {
            Host = host,
            Port = int.TryParse(portStr, out var p) ? p : 1883,
            ClientId = clientId
        };

        logger.LogInformation("[{Channel}] Starting Technical Analysis MQTT client. Host: {Host}, ClientId: {ClientId}", "TechnicalAnalysisChannel", config.Host, config.ClientId);
        await ConnectAsync(config);
    }

    /// <summary>
    /// Stops the MQTT client.
    /// </summary>
    public async Task StopAsync(CancellationToken cancellationToken)
    {
        logger.LogInformation("[{Channel}] Stopping Technical Analysis MQTT client.", "TechnicalAnalysisChannel");
        await DisconnectAsync();
    }

    protected override async Task OnConnectedAsync()
    {
        logger.LogInformation("[{Channel}] Technical Analysis MQTT client connected. Subscribing to RPC topic...", "TechnicalAnalysisChannel");
        await SubscribeAsync("services/request/ta_GetAnalysis/#");
        await SubscribeAsync("services/request/tr_GetLivePrice/#");
        await SubscribeAsync("services/request/health_Ping/#");
        await SubscribeAsync("services/config/updated/#");
    }

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

        if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
        {
            await HandleConfigUpdatedAsync(topic, payload);
            return;
        }

        var segments = topic.Split('/');
        if (segments.Length < 4) return;

        var channel = segments[2];
        var correlationId = segments[segments.Length - 1];

        if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
        {
            await HandleHealthPingAsync(topic, segments, correlationId);
            return;
        }

        if (channel == "ta_GetAnalysis")
        {
            await HandleGetAnalysisAsync(payload, correlationId);
        }
        else if (channel == "tr_GetLivePrice")
        {
            await HandleGetLivePriceAsync(payload, correlationId);
        }
    }

    private async Task HandleConfigUpdatedAsync(string topic, string payload)
    {
        if (!topic.EndsWith("FinlyticTechnicalAnalysis", StringComparison.OrdinalIgnoreCase))
            return;

        logger.LogInformation("[{Channel}] [TAMqttClient] Received config update event for FinlyticTechnicalAnalysis.", "TechnicalAnalysisChannel");
        try
        {
            var updatePayload = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
            if (updatePayload?.Settings != null && updatePayload.Settings.Count > 0)
            {
                using var scope = scopeFactory.CreateScope();
                var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
                await settingsDb.UpdateSettingsFromDictionaryAsync(updatePayload.Settings);
                logger.LogInformation("[{Channel}] [TAMqttClient] Persisted {Count} updated settings to FinlyticTechnicalAnalysis database.", "TechnicalAnalysisChannel", updatePayload.Settings.Count);
            }
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "[{Channel}] [TAMqttClient] Error processing MQTT config update event.", "TechnicalAnalysisChannel");
        }
    }

    private async Task HandleHealthPingAsync(string topic, string[] segments, string correlationId)
    {
        bool isForMe = segments.Length >= 5
            ? segments[3].Equals("FinlyticTechnicalAnalysis", StringComparison.OrdinalIgnoreCase)
            : topic.Contains("FinlyticTechnicalAnalysis", StringComparison.OrdinalIgnoreCase);

        if (isForMe)
        {
            string respTopic = $"services/response/health_Ping/{correlationId}";
            await PublishAsync(respTopic, new FinlyticCore.Dtos.ServiceHealthResponse("FinlyticTechnicalAnalysis", "Online", DateTime.UtcNow, "Connected"));
            logger.LogInformation("[{Channel}] [TAMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "TechnicalAnalysisChannel", correlationId);
        }
    }

    private async Task HandleGetAnalysisAsync(string payload, string correlationId)
    {
        logger.LogInformation("[{Channel}] Received RPC ta_GetAnalysis request. CorrelationId: {CorrelationId}", "TechnicalAnalysisChannel", correlationId);

        var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
        string responseTopic = $"services/response/ta_GetAnalysis/{correlationId}";

        if (string.IsNullOrWhiteSpace(req?.Isin))
        {
            logger.LogWarning("[{Channel}] Request missing mandatory ISIN parameter.", "TechnicalAnalysisChannel");
            await PublishAsync<object?>(responseTopic, null);
            return;
        }

        try
        {
            using var scope = scopeFactory.CreateScope();
            var taDbService = scope.ServiceProvider.GetRequiredService<ITechnicalAnalysisDbService>();

            var analysis = await taDbService.GetAnalysisAsync(req.Isin, req.ForceRefresh, req.Ticker);

            logger.LogInformation("[{Channel}] Publishing RPC response to {ResponseTopic}", "TechnicalAnalysisChannel", responseTopic);
            await PublishAsync(responseTopic, analysis);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "[{Channel}] Failed to fetch technical analysis and publish RPC response for ISIN {Isin}", "TechnicalAnalysisChannel", req.Isin);
            
            // Antworte mit null, damit der Aufrufer nicht im RPC-Timeout verharrt
            try
            {
                await PublishAsync<object?>(responseTopic, null);
            }
            catch { }
        }
    }

    private async Task HandleGetLivePriceAsync(string payload, string correlationId)
    {
        logger.LogInformation("[{Channel}] Received RPC tr_GetLivePrice request. CorrelationId: {CorrelationId}", "TechnicalAnalysisChannel", correlationId);

        var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
        string responseTopic = $"services/response/tr_GetLivePrice/{correlationId}";

        if (string.IsNullOrWhiteSpace(req?.Isin))
        {
            logger.LogWarning("[{Channel}] tr_GetLivePrice request missing mandatory ISIN parameter.", "TechnicalAnalysisChannel");
            await PublishAsync<object?>(responseTopic, null);
            return;
        }

        try
        {
            using var scope = scopeFactory.CreateScope();
            var taDbService = scope.ServiceProvider.GetRequiredService<ITechnicalAnalysisDbService>();

            var livePrice = await taDbService.GetLivePriceAsync(req.Isin);

            logger.LogInformation("[{Channel}] Publishing RPC response to {ResponseTopic} for ISIN {Isin}", "TechnicalAnalysisChannel", responseTopic, req.Isin);
            await PublishAsync(responseTopic, livePrice);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "[{Channel}] Failed to fetch live price and publish RPC response for ISIN {Isin}", "TechnicalAnalysisChannel", req.Isin);

            try
            {
                await PublishAsync<object?>(responseTopic, null);
            }
            catch { }
        }
    }
}