Finlytic/FinlyticAssets/Util/AssetsMqttClient.cs

using System.Text.Json;
using FinlyticAssets.Services;
using FinlyticCore.Entities.Assets;
using FinlyticCore.Models;
using FinlyticCore.Models.Assets;
using FinlyticCore.Util;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace FinlyticAssets.Util;

/// <summary>
/// Represents a managed MQTT client acting as a server-side RPC provider within the asset microservice.
/// It subscribes to request topics, processes incoming JSON payloads via the database and index service, 
/// and publishes the requested asset entities or logo files back to the response topic.
/// Also implements <see cref="IHostedService"/> to manage its own lifecycle connections.
/// </summary>
public class AssetsMqttClient(
    ILogger<AssetsMqttClient> logger, 
    IServiceScopeFactory scopeFactory, 
    IConfiguration configuration) : ManagedMqttClient(logger), IHostedService
{


    /// <summary>
    /// Starts the MQTT client and connects to the configured broker.
    /// </summary>
    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
    /// <returns>A task representing the asynchronous start operation.</returns>
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var config = new MqttConfiguration()
        {
            Host = configuration["MQTT:Host"] ?? configuration["MQTT__Host"]!,
            Port = Convert.ToInt32(configuration["MQTT:Port"] ?? configuration["MQTT__Port"]!),
            ClientId = $"{(configuration["MQTT:ClientId"] ?? configuration["MQTT__ClientId"]!)}_{Guid.NewGuid()}"
        };

        await ConnectAsync(config);
    }

    /// <summary>
    /// Gracefully stops and disconnects the MQTT client.
    /// </summary>
    /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
    /// <returns>A task representing the asynchronous stop operation.</returns>
    public async Task StopAsync(CancellationToken cancellationToken)
    {
        await DisconnectAsync();
    }

    /// <summary>
    /// Invoked automatically once the connection to the MQTT broker is successfully established or restored.
    /// Registers subscriptions for asset validation, search, and logo download requests.
    /// </summary>
    /// <returns>A <see cref="Task"/> representing the asynchronous subscription operation.</returns>
    protected override async Task OnConnectedAsync()
    {
        await SubscribeAsync("services/request/assets_Get/#");
        await SubscribeAsync("services/request/assets_Search/#");
        await SubscribeAsync("services/request/assets_GetDiscovery/#");
        await SubscribeAsync("services/request/assets_FetchLogo/#");
        await SubscribeAsync("services/request/health_Ping/#");
        await SubscribeAsync("services/config/updated/#");
    }


    /// <summary>
    /// Processes incoming messages on the subscribed topics, executes the corresponding service methods, 
    /// and publishes the result to the response topic while preserving the correlation ID.
    /// </summary>
    /// <param name="topic">The MQTT topic on which the message was received.</param>
    /// <param name="payload">The incoming message as a UTF-8 encoded JSON string.</param>
    /// <returns>A <see cref="Task"/> representing the asynchronous message processing operation.</returns>
    protected override async Task OnMessageReceivedAsync(string topic, string payload)
    {
        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;
        }

        try
        {
            using var scope = scopeFactory.CreateScope();
            var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
            var indexService = scope.ServiceProvider.GetRequiredService<IAssetsIndexService>();

            if (channel == "assets_FetchLogo")
            {
                await HandleFetchLogoAsync(payload, correlationId, indexService);
                return;
            }

            List<AssetEntity> responseData = [];

            switch (channel)
            {
                case "assets_Get":
                    responseData = await HandleAssetsGetAsync(payload, dbService);
                    break;
                case "assets_Search":
                    responseData = await HandleAssetsSearchAsync(payload, dbService);
                    break;
                case "assets_GetDiscovery":
                    responseData = await HandleAssetsGetDiscoveryAsync(payload, dbService);
                    break;
            }

            string defaultResponseTopic = $"services/response/{channel}/{correlationId}";
            await PublishAsync(defaultResponseTopic, responseData.ToDtoList());
        }
        catch (Exception ex)
        {
            OnError(ex);
        }
    }

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

        logger.LogInformation("[{Channel}] [AssetsMqttClient] Received config update event for FinlyticAssets.", "AssetsChannel");
        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.UpdateSettingsFromDictionary(updatePayload.Settings);
                logger.LogInformation("[{Channel}] [AssetsMqttClient] Persisted {Count} updated settings to FinlyticAssets database.", "AssetsChannel", updatePayload.Settings.Count);
            }
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "[{Channel}] [AssetsMqttClient] Error processing MQTT config update event.", "AssetsChannel");
        }
    }

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

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

    private async Task HandleFetchLogoAsync(string payload, string correlationId, IAssetsIndexService indexService)
    {
        var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
        string? isin = req?.Isin;
        string? savedPath = null;

        if (!string.IsNullOrEmpty(isin))
        {
            savedPath = await indexService.DownloadAndSaveLogoAsync(isin);
        }

        string responseTopic = $"services/response/assets_FetchLogo/{correlationId}";
        await PublishAsync(responseTopic, new FinlyticCore.Dtos.FetchLogoResponse(isin, savedPath, savedPath != null));
    }

    private async Task<List<AssetEntity>> HandleAssetsGetAsync(string payload, IAssetsDbService dbService)
    {
        var validReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetValidAssetRequest);
        if (validReq != null)
        {
            return await dbService.GetValidAssetsByIsinAsync(validReq.Isin);
        }
        return [];
    }

    private async Task<List<AssetEntity>> HandleAssetsSearchAsync(string payload, IAssetsDbService dbService)
    {
        var searchReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.SearchAssetsRequest);
        if (searchReq != null)
        {
            return await dbService.FindAffectedActiveAssetsAsync(searchReq.SearchQuery);
        }
        return [];
    }

    private async Task<List<AssetEntity>> HandleAssetsGetDiscoveryAsync(string payload, IAssetsDbService dbService)
    {
        int limit = 15;
        if (!string.IsNullOrWhiteSpace(payload))
        {
            try
            {
                var discReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetDiscoveryAssetsRequest);
                if (discReq != null && discReq.Limit > 0) limit = discReq.Limit;
            }
            catch { }
        }
        return await dbService.GetDiscoveryAssetsAsync(limit);
    }
}