Finlytic/FinlyticBot/Util/BotMqttClient.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Dtos;
using FinlyticCore.Dtos.Bot;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Dtos.TechnicalAnalysis;
using FinlyticCore.Dtos.Trading;
using FinlyticCore.Models;
using FinlyticCore.Services;
using FinlyticCore.Util;
using FinlyticBot.Database;
using FinlyticBot.Database.Entities;
using FinlyticBot.Services.Alpaca;
using FinlyticBot.Services.Consumers;
using FinlyticBot.Services.Execution;
using FinlyticBot.Services.Ledger;
using FinlyticBot.Services.Mqtt;
using FinlyticBot.Settings;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace FinlyticBot.Util;

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

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

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

        _logger.LogInformation("Starting FinlyticBot MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
        await _finlyticLogger.LogInfoAsync(BotSettingKeys.MqttChannel, "[BotMqttClient] 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.");
        // Broadcast via IFinlyticLogger too (see OnConnectedAsync's doc comment) so the live console shows a
        // clean "stopped" line instead of just silently going quiet.
        await _finlyticLogger.LogInfoAsync(BotSettingKeys.MqttChannel, "[BotMqttClient] Stopping FinlyticBot MQTT client.");
        await DisconnectAsync();
    }

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

        await SubscribeAsync(MqttTopics.ResponseWildcard);
        await SubscribeRpcAsync<object, BotStatusDto>(MqttTopics.RequestFilter(MqttTopics.Channels.BotGetStatus), HandleGetStatusRpcAsync);
        await SubscribeRpcAsync<object, List<BotTradeOrderDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.BotGetPositions), HandleGetPositionsRpcAsync);
        await SubscribeRpcAsync<object, AccountSummaryDto>(MqttTopics.RequestFilter(MqttTopics.Channels.BotGetSummary), HandleGetSummaryRpcAsync);
        await SubscribeRpcAsync<ExecuteProposalRequest, BotTradeOrderDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.BotExecuteProposal), HandleExecuteProposalRpcAsync);
        await SubscribeRpcAsync<object, PanicCloseResultDto>(MqttTopics.RequestFilter(MqttTopics.Channels.BotPanicClose), HandlePanicCloseRpcAsync);
        await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.BotSettingsGetAll), HandleSettingsGetAllRpcAsync);
        await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.BotSettingsUpdate), HandleSettingsUpdateRpcAsync);
        await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);

        // Subscribe to Engine Proposals
        await SubscribeAsync(MqttTopics.EngineProposalsCreated);

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

        await _finlyticLogger.LogInfoAsync(BotSettingKeys.MqttChannel, "[BotMqttClient] FinlyticBot MQTT client connected. Registering RPC endpoints...");
    }

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

        try
        {
            if (topic.Equals(MqttTopics.EngineProposalsCreated, StringComparison.OrdinalIgnoreCase))
            {
                var proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr, DefaultJsonOptions);
                if (proposal != null && EngineProposalConsumerBackgroundService.OnProposalReceived != null)
                {
                    EngineProposalConsumerBackgroundService.OnProposalReceived(proposal);
                }
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[BotMqttClient] Error handling incoming proposal on topic {Topic}", topic);
        }
    }

    private async Task<BotStatusDto> HandleGetStatusRpcAsync(object? _, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<BotDbContext>();
        var alpacaService = scope.ServiceProvider.GetRequiredService<IAlpacaTradingService>();
        var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();

        int active = await db.Positions.CountAsync(p => p.Status == BotPositionStatus.Active || p.Status == BotPositionStatus.BreakEvenTriggered);
        bool autoExec = await settingsService.GetSettingAsync(BotSettingKeys.EnableAutoExecution);
        int maxPositions = await settingsService.GetSettingAsync(BotSettingKeys.MaxConcurrentPositions);
        decimal riskPerTradePercent = await settingsService.GetSettingAsync(BotSettingKeys.RiskPerTradePercent);

        // MinCompositeScore is owned by FinlyticEngine (EngineSettingKeys.MinCompositeScore ==
        // "Engine.MinCompositeScore", read as `decimal` there — see TradeLifecycleService). FinlyticBot has no
        // project reference to FinlyticEngine (and is out of scope for adding one here), but both services share
        // the same dynamic settings store, so the raw string key is read directly instead of duplicating/inventing
        // a Bot-local setting. Default mirrors EngineSettingKeys' own documented default.
        decimal minCompositeScoreRaw = await settingsService.GetSettingAsync("Engine.MinCompositeScore", 75.0m);

        // The bot worker process itself is always running once started; "IsRunning" from the client's
        // perspective means "is the bot actively acting on proposals", which is exactly what
        // EngineProposalConsumerBackgroundService gates on before calling IBotOrderExecutor. So IsRunning
        // is deliberately the same flag as AutoExecutionEnabled rather than a separate process-alive flag.
        bool isRunning = autoExec;

        // Synthetic broker is always available (in-process ledger); Alpaca is only listed once real
        // credentials are configured (IAlpacaTradingService.IsConfigured), mirroring the venue selection
        // logic in BotOrderExecutor.
        string venuesActive = alpacaService.IsConfigured
            ? $"{BotExecutionVenue.SyntheticPaperBroker}, {BotExecutionVenue.AlpacaPaperTrading}"
            : BotExecutionVenue.SyntheticPaperBroker.ToString();

        return new BotStatusDto(
            IsRunning: isRunning,
            AutoExecutionEnabled: autoExec,
            ActivePositionsCount: active,
            MaxPositions: maxPositions,
            RiskPerTradePercent: riskPerTradePercent,
            MinCompositeScore: (int)Math.Round(minCompositeScoreRaw, MidpointRounding.AwayFromZero),
            VenuesActive: venuesActive
        );
    }

    private async Task<List<BotTradeOrderDto>> HandleGetPositionsRpcAsync(object? _, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<BotDbContext>();

        var positions = await db.Positions
            .AsNoTracking()
            .Where(p => p.Status == BotPositionStatus.Active || p.Status == BotPositionStatus.BreakEvenTriggered || p.Status == BotPositionStatus.Tp1Hit)
            .OrderByDescending(p => p.OpenedAtUtc)
            .ToListAsync();

        return positions.Select(BotOrderExecutor.MapEntityToDto).ToList();
    }

    private async Task<AccountSummaryDto> HandleGetSummaryRpcAsync(object? _, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var syntheticBroker = scope.ServiceProvider.GetRequiredService<ISyntheticPaperBroker>();
        return await syntheticBroker.GetSummaryAsync();
    }

    private async Task<BotTradeOrderDto?> HandleExecuteProposalRpcAsync(ExecuteProposalRequest? req, string correlationId)
    {
        if (req == null) return null;

        using var scope = _scopeFactory.CreateScope();
        var executor = scope.ServiceProvider.GetRequiredService<IBotOrderExecutor>();
        var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();

        // Look up the existing proposal by its actual GUID via the engine_GetProposals RPC channel — the
        // same channel EngineController/UserTradesController already use for proposal listings.
        // (req.ProposalId is a proposal GUID, not an ISIN; a previous version of this method sent it to
        // engine_EvaluateIsin as if it were one, which started a full re-evaluation of a nonsensical "ISIN"
        // and could never resolve a proposal — see Bug A of the security review.)
        // A generous limit is used because the proposal the caller wants to execute may not be among the
        // most recently created handful if several proposals were generated in quick succession.
        List<TradeProposalDto>? proposals;
        try
        {
            proposals = await SendRpcRequestAsync<List<TradeProposalDto>, GetTradeProposalsRequest>(
                MqttTopics.Channels.EngineGetProposals,
                new GetTradeProposalsRequest(OnlyActive: true, Limit: 200),
                TimeSpan.FromSeconds(5)
            );
        }
        catch (Exception ex)
        {
            await logger.LogWarningAsync(BotSettingKeys.BotChannel, ex,
                "[BotMqttClient] RPC failure while fetching proposals from FinlyticEngine to resolve proposal {ProposalId} for execution [CorrelationId: {CorrelationId}]",
                req.ProposalId, correlationId);
            return null;
        }

        if (proposals == null)
        {
            await logger.LogWarningAsync(BotSettingKeys.BotChannel,
                "[BotMqttClient] FinlyticEngine did not respond to engine_GetProposals in time while resolving proposal {ProposalId} for execution [CorrelationId: {CorrelationId}]",
                req.ProposalId, correlationId);
            return null;
        }

        var proposal = proposals.FirstOrDefault(p => p.ProposalId == req.ProposalId);
        if (proposal == null)
        {
            await logger.LogWarningAsync(BotSettingKeys.BotChannel,
                "[BotMqttClient] Proposal {ProposalId} was not found among FinlyticEngine's active proposals (expired or invalid) [CorrelationId: {CorrelationId}]",
                req.ProposalId, correlationId);
            return null;
        }

        await logger.LogInfoAsync(BotSettingKeys.BotChannel,
            "[BotMqttClient] Executing manual paper trade for proposal {Id} [CorrelationId: {CorrelationId}]", req.ProposalId, correlationId);
        return await executor.ExecuteProposalAsync(proposal, req.PreferredVenue, req.CustomQuantity);
    }

    /// <summary>
    /// Emergency-closes every currently open paper-trading position (Status in Active, BreakEvenTriggered,
    /// Tp1Hit, Tp2Hit — the terminal statuses Closed/StoppedOut/KnockedOut/Canceled are, by definition,
    /// already not open and Pending is not currently assigned by any code path).
    /// <para>
    /// Synthetic ledger positions are closed unconditionally via <see cref="ISyntheticPaperBroker"/> — there
    /// is no external broker to confirm with, so the internal ledger IS the authority.
    /// </para>
    /// <para>
    /// Alpaca positions are the safety-critical case: this handler NEVER marks an Alpaca position as closed
    /// unless <see cref="IAlpacaTradingService.ClosePositionAsync"/> returns successfully (i.e. Alpaca's REST
    /// API confirmed it accepted the liquidation order). If Alpaca is not configured, or the broker call
    /// throws, the position is left completely untouched in the database and is counted in
    /// <see cref="PanicCloseResultDto.SkippedCount"/> rather than silently reported as closed
    /// (Rules.md §4 — a false-positive "closed" on a real broker position would be fatal).
    /// </para>
    /// </summary>
    private async Task<PanicCloseResultDto> HandlePanicCloseRpcAsync(object? _, string correlationId)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<BotDbContext>();
        var syntheticBroker = scope.ServiceProvider.GetRequiredService<ISyntheticPaperBroker>();
        var alpacaService = scope.ServiceProvider.GetRequiredService<IAlpacaTradingService>();
        var logger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<BotMqttClient>>();

        var openStatuses = new[]
        {
            BotPositionStatus.Active,
            BotPositionStatus.BreakEvenTriggered,
            BotPositionStatus.Tp1Hit,
            BotPositionStatus.Tp2Hit
        };

        var positions = await db.Positions
            .Where(p => openStatuses.Contains(p.Status))
            .OrderBy(p => p.OpenedAtUtc)
            .ToListAsync();

        await logger.LogWarningAsync(BotSettingKeys.BotChannel,
            "[BotMqttClient] PANIC CLOSE triggered: attempting to close {Count} open position(s) [CorrelationId: {CorrelationId}]",
            positions.Count, correlationId);

        var closedOrders = new List<BotTradeOrderDto>();
        int skippedCount = 0;

        foreach (var pos in positions)
        {
            if (pos.Venue == BotExecutionVenue.SyntheticPaperBroker)
            {
                // No external broker to confirm with — the internal ledger is the authority for its own
                // positions, so this closes unconditionally at the last synced price (same direction-aware
                // realized P&L formula ISyntheticPaperBroker already uses elsewhere for consistency).
                var closed = await syntheticBroker.ClosePositionAsync(pos.Id, pos.CurrentPrice, BotPositionStatus.Closed);
                closedOrders.Add(BotOrderExecutor.MapEntityToDto(closed));
                continue;
            }

            // Alpaca venue: only ever mark closed after the broker confirms the liquidation.
            if (!alpacaService.IsConfigured)
            {
                await logger.LogWarningAsync(BotSettingKeys.BotChannel,
                    "[BotMqttClient] PANIC CLOSE SKIPPED Alpaca position {PositionId} ({Symbol}): Alpaca is not configured. Position left OPEN in the ledger — manual intervention required.",
                    pos.Id, pos.Symbol);
                skippedCount++;
                continue;
            }

            try
            {
                var closeResult = await alpacaService.ClosePositionAsync(pos.Symbol);

                // Alpaca confirmed acceptance of the liquidation order — only now is it safe to persist
                // the position as closed. AverageFillPrice can still be null immediately after submission
                // (e.g. outside market hours); fall back to the last synced price rather than fabricating one.
                decimal exitPrice = closeResult.AverageFillPrice ?? pos.CurrentPrice;
                pos.Status = BotPositionStatus.Closed;
                pos.ClosedAtUtc = DateTime.UtcNow;
                pos.CurrentPrice = exitPrice;
                pos.LastSyncAtUtc = DateTime.UtcNow;
                pos.RealizedPnlEur = CalculateRealizedPnl(pos, exitPrice);

                await db.SaveChangesAsync();

                await logger.LogInfoAsync(BotSettingKeys.BotChannel,
                    "[BotMqttClient] PANIC CLOSE confirmed by Alpaca for position {PositionId} ({Symbol}): Order {OrderId} (Status: {Status}), exit {Exit:F2} €.",
                    pos.Id, pos.Symbol, closeResult.OrderId, closeResult.OrderStatus, exitPrice);

                closedOrders.Add(BotOrderExecutor.MapEntityToDto(pos));
            }
            catch (Exception ex)
            {
                // The broker call itself failed (not configured mid-flight, network/HTTP error, or Alpaca
                // rejected the request) — the position is left completely untouched, exactly as it was
                // before this handler ran. It must NOT be counted as closed.
                await logger.LogWarningAsync(BotSettingKeys.BotChannel, ex,
                    "[BotMqttClient] PANIC CLOSE FAILED for Alpaca position {PositionId} ({Symbol}): broker call did not confirm liquidation. Position left OPEN in the ledger — manual intervention required.",
                    pos.Id, pos.Symbol);
                skippedCount++;
            }
        }

        return new PanicCloseResultDto(closedOrders.Count, skippedCount, closedOrders);
    }

    /// <summary>
    /// Direction-aware realized P&amp;L for a position closed at <paramref name="exitPrice"/>, mirroring the
    /// formula <see cref="ISyntheticPaperBroker"/> already uses for its own (non-knock-out) closes, so both
    /// venues report P&amp;L consistently. Only used for the Alpaca panic-close path here — the synthetic path
    /// delegates to <see cref="ISyntheticPaperBroker.ClosePositionAsync"/> directly, which computes its own.
    /// </summary>
    private static decimal CalculateRealizedPnl(BotPositionEntity pos, decimal exitPrice)
    {
        decimal pnl = pos.Direction == SignalDirection.Buy
            ? ((exitPrice - pos.AverageBuyIn) * pos.Quantity) - pos.TotalFeesEur
            : ((pos.AverageBuyIn - exitPrice) * pos.Quantity) - pos.TotalFeesEur;

        return Math.Round(pnl, 2);
    }

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

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

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

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