Finlytic/FinlyticNotify/Services/NotifyMqttClient.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.News;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Dtos.Trading;
using FinlyticCore.Models;
using FinlyticCore.Services;
using FinlyticCore.Util;
using FinlyticNotify.Settings;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace FinlyticNotify.Services;

/// <summary>
/// Managed MQTT client for FinlyticNotify. Listens strictly to existing MQTT broadcast topics,
/// resolves trade ownership, and dispatches rich push notifications via ntfy.
/// </summary>
public class NotifyMqttClient : ManagedMqttClient, IHostedService
{
    private readonly IConfiguration _configuration;
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly INtfyClient _ntfyClient;
    private readonly INotificationFormatter _formatter;
    private readonly IUserTradeResolver _userTradeResolver;
    private readonly ISettingsService? _settingsService;
    private readonly IFinlyticLogger<NotifyMqttClient>? _finlyticLogger;
    private readonly ILogger<NotifyMqttClient> _logger;

    /// <summary>
    /// Initializes a new instance of the <see cref="NotifyMqttClient"/> class.
    /// </summary>
    public NotifyMqttClient(
        IConfiguration configuration,
        IServiceScopeFactory scopeFactory,
        INtfyClient ntfyClient,
        INotificationFormatter formatter,
        IUserTradeResolver userTradeResolver,
        ILogger<NotifyMqttClient> logger,
        ISettingsService? settingsService = null,
        IFinlyticLogger<NotifyMqttClient>? finlyticLogger = null) : base(logger)
    {
        _configuration = configuration;
        _scopeFactory = scopeFactory;
        _ntfyClient = ntfyClient;
        _formatter = formatter;
        _userTradeResolver = userTradeResolver;
        _settingsService = settingsService;
        _finlyticLogger = finlyticLogger;
        _logger = logger;
    }

    /// <inheritdoc />
    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticNotify");
        _logger.LogInformation("[NotifyMqttClient] Starting FinlyticNotify MQTT client (Broker: {Host}:{Port}, ClientId: {ClientId})",
            config.Host, config.Port, config.ClientId);

        await ConnectAsync(config);
    }

    /// <inheritdoc />
    public async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("[NotifyMqttClient] Stopping FinlyticNotify MQTT client.");
        if (_finlyticLogger != null)
        {
            await _finlyticLogger.LogInfoAsync(NotifySettingKeys.MqttChannel, "[NotifyMqttClient] Stopping FinlyticNotify MQTT client.");
        }
        await DisconnectAsync();
    }

    /// <inheritdoc />
    protected override async Task OnConnectedAsync()
    {
        _logger.LogInformation("[NotifyMqttClient] Connected to MQTT broker. Subscribing to trade event topics...");

        await SubscribeAsync(MqttTopics.ResponseWildcard);

        // 1. Subscribe to Trade Proposals (New Trades / Setups)
        await SubscribeAsync(MqttTopics.EngineProposalsCreated);

        // 2. Subscribe to Trade Lifecycle Status Changes (Fills, SL-Updates, TPs, Exits)
        await SubscribeAsync(MqttTopics.EngineTradesStatusChanged);

        // 3. Subscribe to Bot Paper-Trading Streams
        await SubscribeAsync(MqttTopics.BotTradesStream);

        // 4. Subscribe to News Status Update requests to capture analyzed news events
        await SubscribeAsync<UpdateNewsStatusRequest>(
            MqttTopics.RequestFilter(MqttTopics.Channels.NewsUpdateStatus), HandleNewsStatusUpdateRequestAsync);

        // 5. Subscribe to Service Health Ping for fleet monitoring
        await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingTopicAsync);

        // 6. Subscribe to Dynamic Settings RPC Channels
        await SubscribeRpcAsync<object, List<DynamicSettingDto>>(
            MqttTopics.RequestFilter(MqttTopics.Channels.NotifySettingsGetAll), HandleSettingsGetAllRpcAsync);
        await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(
            MqttTopics.RequestFilter(MqttTopics.Channels.NotifySettingsUpdate), HandleSettingsUpdateRpcAsync);

        // 7. Wire structured log broadcasting over MQTT
        FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
        {
            if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticNotify", StringComparison.OrdinalIgnoreCase))
            {
                await PublishAsync(MqttTopics.Logs("FinlyticNotify"), logDto);
            }
        };

        if (_finlyticLogger != null)
        {
            await _finlyticLogger.LogInfoAsync(NotifySettingKeys.MqttChannel,
                "[NotifyMqttClient] FinlyticNotify MQTT client connected and subscribed to trade and news events.");
        }
    }

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

        string topicPrefix = "finlytic";
        if (_settingsService != null)
        {
            topicPrefix = await _settingsService.GetSettingAsync(NotifySettingKeys.NtfyTopicPrefix);
        }
        else
        {
            topicPrefix = _configuration.GetValue<string>("Ntfy:TopicPrefix") ?? NotifySettingKeys.NtfyTopicPrefix.DefaultValue;
        }

        topicPrefix = topicPrefix.Trim('/');

        try
        {
            // Case A: New Trade Proposals created by FinlyticEngine
            if (topic.Equals(MqttTopics.EngineProposalsCreated, StringComparison.OrdinalIgnoreCase))
            {
                await HandleProposalCreatedAsync(payloadStr, topicPrefix);
            }
            // Case B: Trade Status Changed (Lifecycle updates for active trades)
            else if (topic.Equals(MqttTopics.EngineTradesStatusChanged, StringComparison.OrdinalIgnoreCase))
            {
                await HandleTradeStatusChangedAsync(payloadStr, topicPrefix);
            }
            // Case C: Bot Paper-Trading Execution stream
            else if (topic.Equals(MqttTopics.BotTradesStream, StringComparison.OrdinalIgnoreCase))
            {
                await HandleBotTradeStreamAsync(payloadStr, topicPrefix);
            }
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[NotifyMqttClient] Unexpected error handling message on topic {Topic}", topic);
            if (_finlyticLogger != null)
            {
                await _finlyticLogger.LogErrorAsync(NotifySettingKeys.NotifyChannel, ex,
                    "[NotifyMqttClient] Unexpected error handling message on topic {Topic}", topic);
            }
        }
    }

    private async Task HandleProposalCreatedAsync(string payloadStr, string topicPrefix)
    {
        bool notifyOnProposals = true;
        decimal minScore = 70.0m;
        string broadcastChannel = "broadcast";

        if (_settingsService != null)
        {
            notifyOnProposals = await _settingsService.GetSettingAsync(NotifySettingKeys.NotifyOnProposals);
            minScore = await _settingsService.GetSettingAsync(NotifySettingKeys.NtfyMinProposalScore);
            broadcastChannel = await _settingsService.GetSettingAsync(NotifySettingKeys.NtfyBroadcastChannel);
        }
        else
        {
            notifyOnProposals = _configuration.GetValue<bool>("Ntfy:NotifyOnProposals", true);
            minScore = _configuration.GetValue<decimal>("Ntfy:MinProposalScore", 70.0m);
            broadcastChannel = _configuration.GetValue<string>("Ntfy:BroadcastChannel") ?? "broadcast";
        }

        if (!notifyOnProposals) return;

        var proposal = JsonSerializer.Deserialize<TradeProposalDto>(payloadStr, DefaultJsonOptions);
        if (proposal == null) return;

        if (proposal.CompositeScore < minScore)
        {
            _logger.LogDebug("[NotifyMqttClient] Skipping proposal {ProposalId}: CompositeScore {Score} < MinScore {MinScore}",
                proposal.ProposalId, proposal.CompositeScore, minScore);
            return;
        }

        string targetTopic = $"{topicPrefix}_{broadcastChannel}";
        var notification = _formatter.FormatProposalNotification(proposal, targetTopic);
        await _ntfyClient.SendNotificationAsync(notification);
    }

    private async Task HandleTradeStatusChangedAsync(string payloadStr, string topicPrefix)
    {
        bool notifyOnTradeUpdates = true;
        if (_settingsService != null)
        {
            notifyOnTradeUpdates = await _settingsService.GetSettingAsync(NotifySettingKeys.NotifyOnTradeUpdates);
        }
        else
        {
            notifyOnTradeUpdates = _configuration.GetValue<bool>("Ntfy:NotifyOnTradeUpdates", true);
        }

        if (!notifyOnTradeUpdates) return;

        var trade = JsonSerializer.Deserialize<ActiveTradeDto>(payloadStr, DefaultJsonOptions);
        if (trade == null) return;

        // Resolve which user owns this trade
        string username = await _userTradeResolver.ResolveUsernameByUserIdAsync(trade.UserId);
        string targetTopic = $"{topicPrefix}_{username}";

        var notification = _formatter.FormatTradeStatusNotification(trade, targetTopic);
        await _ntfyClient.SendNotificationAsync(notification);
    }

    private async Task HandleBotTradeStreamAsync(string payloadStr, string topicPrefix)
    {
        bool notifyOnBotTrades = true;
        if (_settingsService != null)
        {
            notifyOnBotTrades = await _settingsService.GetSettingAsync(NotifySettingKeys.NotifyOnBotTrades);
        }
        else
        {
            notifyOnBotTrades = _configuration.GetValue<bool>("Ntfy:NotifyOnBotTrades", true);
        }

        if (!notifyOnBotTrades) return;

        var botTrade = JsonSerializer.Deserialize<BotTradeOrderDto>(payloadStr, DefaultJsonOptions);
        if (botTrade == null) return;

        string targetTopic = $"{topicPrefix}_bot";
        var notification = _formatter.FormatBotTradeNotification(botTrade, targetTopic);
        await _ntfyClient.SendNotificationAsync(notification);
    }

    private async Task HandleNewsStatusUpdateRequestAsync(UpdateNewsStatusRequest? req, string topic, string correlationId)
    {
        if (req == null || req.Id == Guid.Empty) return;

        // Only trigger push notifications when an article's status is transitioning to "Analyzed"
        if (!string.Equals(req.Status, "Analyzed", StringComparison.OrdinalIgnoreCase)) return;

        bool notifyOnNews = true;
        string newsChannel = "news";
        string topicPrefix = "finlytic";

        if (_settingsService != null)
        {
            notifyOnNews = await _settingsService.GetSettingAsync(NotifySettingKeys.NotifyOnNews);
            newsChannel = await _settingsService.GetSettingAsync(NotifySettingKeys.NtfyNewsChannel);
            topicPrefix = await _settingsService.GetSettingAsync(NotifySettingKeys.NtfyTopicPrefix);
        }
        else
        {
            notifyOnNews = _configuration.GetValue<bool>("Ntfy:NotifyOnNews", true);
            newsChannel = _configuration.GetValue<string>("Ntfy:NewsChannel") ?? "news";
            topicPrefix = _configuration.GetValue<string>("Ntfy:TopicPrefix") ?? "finlytic";
        }

        if (!notifyOnNews) return;

        try
        {
            // Query FinlyticNews via existing news_GetById RPC channel to get the full enriched NewsArticleDto
            var article = await SendRpcRequestAsync<NewsArticleDto, ArticleRequest>(
                MqttTopics.Channels.NewsGetById,
                new ArticleRequest(req.Id.ToString(), req.Id.ToString()),
                TimeSpan.FromSeconds(5));

            if (article != null)
            {
                string targetTopic = $"{topicPrefix.Trim('/')}_{newsChannel}";
                var notification = _formatter.FormatNewsNotification(article, targetTopic);
                await _ntfyClient.SendNotificationAsync(notification);
            }
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[NotifyMqttClient] Failed to fetch analyzed news article {ArticleId} via news_GetById RPC.", req.Id);
        }
    }

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

            if (_finlyticLogger != null)
            {
                await _finlyticLogger.LogInfoAsync(NotifySettingKeys.HealthPingChannel,
                    "[FinlyticNotify] Responded to live health_Ping RPC [CorrelationId: {CorrelationId}].", correlationId);
            }
        }
    }

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

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