Finlytic/FinlyticBot/Util/BotMqttClient.cs

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

namespace FinlyticBot.Util;

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

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

    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"),
            Username = _configuration["MQTT:Username"] ?? _configuration["MQTT__Username"],
            Password = _configuration["MQTT:Password"] ?? _configuration["MQTT__Password"],
            ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "finlytic_bot")}_{Guid.NewGuid():N}"
        };

        _logger.LogInformation("Starting FinlyticBot MQTT Client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
        await ConnectAsync(config);
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Stopping FinlyticBot MQTT Client.");
        await DisconnectAsync();
    }

    protected override async Task OnConnectedAsync()
    {
        _logger.LogInformation("FinlyticBot MQTT Client connected. Subscribing to topics...");

        await SubscribeAsync("services/events/trades/proposal");
        await SubscribeAsync("finlytic/trades/proposed/#");
        await SubscribeAsync("services/events/analyzer/trade_proposed");
        await SubscribeAsync("services/request/bot_Settings_GetAll/#");
        await SubscribeAsync("services/request/bot_Settings_Update/#");
        await SubscribeAsync("services/request/health_Ping/#");

        FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
        {
            if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticBot", StringComparison.OrdinalIgnoreCase))
            {
                await PublishAsync("finlytic/logs/FinlyticBot", logDto);
            }
        };

        _logger.LogInformation("Successfully subscribed to FinlyticBot event and RPC channels.");
    }

    protected override async Task OnMessageReceivedAsync(string topic, string payloadStr)
    {
        try
        {
            if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
            {
                var segments = topic.Split('/');
                bool isForMe = segments.Length >= 5
                    ? segments[3].Equals("FinlyticBot", StringComparison.OrdinalIgnoreCase)
                    : topic.Contains("FinlyticBot", StringComparison.OrdinalIgnoreCase);

                if (isForMe)
                {
                    string correlationId = segments[^1];
                    string respTopic = $"services/response/health_Ping/{correlationId}";
                    await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticBot", "Online", DateTime.UtcNow, "Connected"));

                    using var scope = _scopeFactory.CreateScope();
                    var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();
                    await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel,
                        "[FinlyticBot] Responded to live health_Ping RPC [CorrelationId: {CorrelationId}].", correlationId);
                }
                return;
            }

            if (topic.StartsWith("services/request/bot_Settings_GetAll/", StringComparison.OrdinalIgnoreCase))
            {
                var correlationId = topic.Split('/')[^1];
                await HandleSettingsGetAllAsync(correlationId);
                return;
            }

            if (topic.StartsWith("services/request/bot_Settings_Update/", StringComparison.OrdinalIgnoreCase))
            {
                var correlationId = topic.Split('/')[^1];
                await HandleSettingsUpdateAsync(correlationId, payloadStr);
                return;
            }

            if (topic.Equals("services/events/trades/proposal", StringComparison.OrdinalIgnoreCase) ||
                topic.StartsWith("finlytic/trades/proposed/", StringComparison.OrdinalIgnoreCase) ||
                topic.Equals("services/events/analyzer/trade_proposed", StringComparison.OrdinalIgnoreCase))
            {
                await HandleTradeProposalEventAsync(payloadStr);
                return;
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error processing incoming MQTT message on topic {Topic}", topic);
        }
    }

    private async Task HandleTradeProposalEventAsync(string payloadStr)
    {
        if (string.IsNullOrWhiteSpace(payloadStr)) return;

        TradeProposalDto? proposal = null;
        try
        {
            proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr, FinlyticJsonSerializerContext.Default.TradeProposalDto);
        }
        catch
        {
            proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr);
        }

        if (proposal == null) return;

        using var scope = _scopeFactory.CreateScope();
        var executionService = scope.ServiceProvider.GetRequiredService<IBotOrderExecutionService>();
        await executionService.ProcessTradeProposalAsync(proposal);
    }

    private async Task HandleSettingsGetAllAsync(string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
        var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });

        var responseTopic = $"services/response/bot_Settings_GetAll/{correlationId}";
        await PublishAsync(responseTopic, settings);
    }

    private async Task HandleSettingsUpdateAsync(string correlationId, string payload)
    {
        using var scope = _scopeFactory.CreateScope();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();

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

        var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
        var responseTopic = $"services/response/bot_Settings_Update/{correlationId}";
        await PublishAsync(responseTopic, currentSettings);
    }
}