Finlytic/FinlyticEngine/Util/EngineMqttClient.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Dtos;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Dtos.Trading;
using FinlyticCore.Models;
using FinlyticCore.Services;
using FinlyticCore.Util;
using FinlyticEngine.Services.Mqtt;
using FinlyticEngine.Services.Trading;
using FinlyticEngine.Settings;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace FinlyticEngine.Util;

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

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

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticEngine");

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

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

    protected override async Task OnConnectedAsync()
    {
        _logger.LogInformation("FinlyticEngine MQTT client connected. Registering RPC endpoints...");

        await SubscribeAsync(MqttTopics.ResponseWildcard);
        await SubscribeRpcAsync<GetTradeProposalsRequest, List<TradeProposalDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineGetProposals), HandleGetProposalsRpcAsync);
        await SubscribeRpcAsync<GetActiveTradesRequest, List<ActiveTradeDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineGetTrades), HandleGetTradesRpcAsync);
        await SubscribeRpcAsync<EvaluateAssetRequest, AssetEvaluationResultDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineEvaluateIsin), HandleEvaluateIsinRpcAsync);
        await SubscribeRpcAsync<GetEvaluationHistoryRequest, GetEvaluationHistoryResponse>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineGetEvaluationHistory), HandleGetEvaluationHistoryRpcAsync);
        await SubscribeRpcAsync<AddTradeFillRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineAddFill), HandleAddFillRpcAsync);
        await SubscribeRpcAsync<UpdateTradeStopLossRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineUpdateStopLoss), HandleUpdateStopLossRpcAsync);
        await SubscribeRpcAsync<CloseEngineTradeRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineCloseTrade), HandleCloseTradeRpcAsync);
        await SubscribeRpcAsync<AcceptTradeProposalRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineAcceptProposal), HandleAcceptProposalRpcAsync);
        await SubscribeRpcAsync<CreateManualTradeRequest, ActiveTradeDto>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineCreateManualTrade), HandleCreateManualTradeRpcAsync);
        await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineSettingsGetAll), HandleSettingsGetAllRpcAsync);
        await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.EngineSettingsUpdate), HandleSettingsUpdateRpcAsync);
        await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);

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

    private async Task<List<TradeProposalDto>> HandleGetProposalsRpcAsync(GetTradeProposalsRequest? req, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.GetProposalsAsync(req?.OnlyActive ?? true, req?.Limit ?? 50);
    }

    private async Task<List<ActiveTradeDto>> HandleGetTradesRpcAsync(GetActiveTradesRequest? req, string correlationId)
    {
        if (req == null) throw new ArgumentNullException(nameof(req));

        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.GetActiveTradesAsync(req.UserId, req.Mode);
    }

    private async Task<AssetEvaluationResultDto> HandleEvaluateIsinRpcAsync(EvaluateAssetRequest? req, string correlationId)
    {
        // A blank/missing ISIN is no longer a special case here: EvaluateAssetAsync now always returns a
        // populated AssetEvaluationResultDto (never null), including for a blank ISIN, so it is safe to just
        // delegate straight through.
        //
        // This RPC channel is only ever reached from the manual, on-demand Web UI flows
        // (AnalyzeController.TriggerManualAnalysis / EngineController.EvaluateAsset) - the autonomous
        // OpportunityPollerBackgroundService calls ITradeLifecycleService.EvaluateAssetAsync directly
        // in-process and never goes through MQTT for it - so TriggerSource is always Manual here. UserId comes
        // from EvaluateAssetRequest.UserId, which FinlyticBackend always overwrites server-side with the JWT
        // identity before publishing the request (see EvaluateAssetRequest's doc comment); Guid.Empty (the
        // request's own default) is treated as "no identity available" rather than a real user ID.
        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        Guid? triggeredByUserId = req != null && req.UserId != Guid.Empty ? req.UserId : null;
        return await lifecycleService.EvaluateAssetAsync(
            req?.Isin ?? string.Empty, req?.Ticker, req?.ForceAiEvaluation ?? false,
            triggerSource: TriggerSource.Manual, triggeredByUserId: triggeredByUserId);
    }

    private async Task<GetEvaluationHistoryResponse> HandleGetEvaluationHistoryRpcAsync(GetEvaluationHistoryRequest? req, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var historyService = scope.ServiceProvider.GetRequiredService<IEvaluationHistoryService>();
        return await historyService.GetHistoryAsync(req ?? new GetEvaluationHistoryRequest());
    }

    private async Task<ActiveTradeDto> HandleAddFillRpcAsync(AddTradeFillRequest? req, string correlationId)
    {
        if (req == null) throw new ArgumentNullException(nameof(req));

        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.AddTradeFillAsync(req.UserId, req.TradeId, req.ExecutedPrice, req.Quantity, req.Fee, req.Note);
    }

    private async Task<ActiveTradeDto> HandleUpdateStopLossRpcAsync(UpdateTradeStopLossRequest? req, string correlationId)
    {
        if (req == null) throw new ArgumentNullException(nameof(req));

        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.UpdateStopLossAsync(req.UserId, req.TradeId, req.NewStopLoss, req.Reason);
    }

    private async Task<ActiveTradeDto> HandleCloseTradeRpcAsync(CloseEngineTradeRequest? req, string correlationId)
    {
        if (req == null) throw new ArgumentNullException(nameof(req));

        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.CloseTradeAsync(req.UserId, req.TradeId, req.ClosePrice, req.Reason);
    }

    private async Task<ActiveTradeDto> HandleAcceptProposalRpcAsync(AcceptTradeProposalRequest? req, string correlationId)
    {
        if (req == null) throw new ArgumentNullException(nameof(req));

        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.AcceptProposalAsync(req);
    }

    private async Task<ActiveTradeDto> HandleCreateManualTradeRpcAsync(CreateManualTradeRequest? req, string correlationId)
    {
        if (req == null) throw new ArgumentNullException(nameof(req));

        using var scope = _scopeFactory.CreateScope();
        var lifecycleService = scope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
        return await lifecycleService.CreateManualTradeAsync(req);
    }

    private async Task<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
        return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(EngineSettingKeys) });
    }

    private async Task<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();

        if (updates != null && updates.Count > 0)
        {
            await settingsService.UpdateSettingsAsync(updates);
        }

        return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(EngineSettingKeys) });
    }

    private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
    {
        if (topic.Contains("FinlyticEngine", StringComparison.OrdinalIgnoreCase) || !topic.Contains("/", StringComparison.OrdinalIgnoreCase))
        {
            string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
            await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticEngine", "Online", DateTime.UtcNow, "Connected"));

            using var scope = _scopeFactory.CreateScope();
            var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<EngineMqttClient>>();
            await logger.LogInfoAsync(EngineSettingKeys.HealthPingChannel,
                "[FinlyticEngine] Responded to health_Ping RPC [CorrelationId: {CorrelationId}]", correlationId);
        }
    }
}