Finlytic/FinlyticCore/Util/ManagedMqttClient.cs

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Dtos;
using FinlyticCore.Models;
using FinlyticCore.Models.Settings;
using FinlyticCore.Services;
using Microsoft.Extensions.Logging;
using MQTTnet;

namespace FinlyticCore.Util;

/// <summary>
/// An abstract, resilient MQTT client wrapper designed for microservice architectures.
/// Handles automatic reconnection, structured JSON publishing, thread-safe subscription management,
/// typed/generic message handling, and synchronous Request-Reply (RPC).
/// Supports channel-controlled logging via <see cref="CoreSettingKeys.MqttChannel"/>.
/// </summary>
public abstract class ManagedMqttClient : IDisposable
{
    protected static readonly JsonSerializerOptions DefaultJsonOptions = new()
    {
        PropertyNameCaseInsensitive = true,
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
    };

    private readonly ILogger<ManagedMqttClient> _logger;
    private readonly ISettingsService? _settingsService;
    private readonly IFinlyticLogger<ManagedMqttClient>? _finlyticLogger;
    private readonly IMqttClient _mqttClient;
    private CancellationTokenSource? _cts;
    
    // Tracks pending RPC requests waiting for a specific correlation ID reply
    private readonly ConcurrentDictionary<string, TaskCompletionSource<string>> _pendingRequests = new();

    /// <summary>
    /// Literal suffix appended to a normal RPC response topic to build its "fault" sibling topic, e.g.
    /// <c>services/response/{channel}/{correlationId}/error</c>. Publishing faults on a distinct topic (instead
    /// of on the regular response topic with some in-payload error marker) lets a caller recognize a fault
    /// deterministically from the topic string alone, before ever attempting to deserialize the body as the
    /// expected <c>TResponse</c> — which matters because a generic RPC client has no way to heuristically tell a
    /// legitimate <c>TResponse</c> payload apart from an error payload shaped like something else.
    /// It also makes the scheme degrade safely across a rolling deployment: an old client (pre-dating this
    /// suffix) that receives a new server's fault message extracts "error" as a bogus correlation ID, finds no
    /// matching pending request, and simply falls through — it keeps waiting and eventually times out exactly as
    /// it did before this feature existed, instead of crashing or misinterpreting the payload. Symmetrically, a
    /// new client talking to an old server that never publishes this topic at all simply times out as before.
    /// </summary>
    private const string ErrorTopicSuffix = "/error";

    // Tracks registered topic handlers for direct routing
    private readonly ConcurrentDictionary<string, List<Func<string, string, Task>>> _topicHandlers = new(StringComparer.OrdinalIgnoreCase);

    // Tracks all active topic filters for automatic re-subscription on reconnect
    private readonly ConcurrentDictionary<string, bool> _subscribedTopics = new(StringComparer.OrdinalIgnoreCase);

    /// <summary>
    /// Gets a value indicating whether the client is currently connected to the MQTT broker.
    /// </summary>
    public bool IsConnected => _mqttClient.IsConnected;

    protected ManagedMqttClient(
        ILogger<ManagedMqttClient> logger,
        ISettingsService? settingsService = null,
        IFinlyticLogger<ManagedMqttClient>? finlyticLogger = null)
    {
        _logger = logger;
        _settingsService = settingsService;
        _finlyticLogger = finlyticLogger;
        _mqttClient = new MqttClientFactory().CreateMqttClient();
        
        _mqttClient.ApplicationMessageReceivedAsync += HandleIncomingMessageAsync;
        _mqttClient.DisconnectedAsync += HandleDisconnectAsync;
    }

    private async Task LogMqttInfoAsync(string message, params object[] args)
    {
        if (_finlyticLogger != null)
        {
            await _finlyticLogger.LogInfoAsync(CoreSettingKeys.MqttChannel, message, args);
        }
        else if (_settingsService != null)
        {
            if (await _settingsService.GetSettingAsync(CoreSettingKeys.MqttChannel))
            {
                _logger.LogInformation(message, args);
            }
        }
        else
        {
            _logger.LogInformation(message, args);
        }
    }

    private async Task LogMqttDebugAsync(string message, params object[] args)
    {
        if (_finlyticLogger != null)
        {
            await _finlyticLogger.LogDebugAsync(CoreSettingKeys.MqttChannel, message, args);
        }
        else if (_settingsService != null)
        {
            if (await _settingsService.GetSettingAsync(CoreSettingKeys.MqttChannel))
            {
                _logger.LogDebug(message, args);
            }
        }
        else
        {
            _logger.LogDebug(message, args);
        }
    }

    /// <summary>
    /// Establishes a connection to the MQTT broker and initializes the background auto-reconnection loop.
    /// </summary>
    public async Task ConnectAsync(MqttConfiguration config)
    {
        if (IsConnected)
            throw new InvalidOperationException("MQTT client is already connected.");

        _cts = new CancellationTokenSource();

        var optionsBuilder = new MqttClientOptionsBuilder()
            .WithTcpServer(config.Host, config.Port)
            .WithClientId(config.ClientId)
            .WithCleanSession();

        if (!string.IsNullOrWhiteSpace(config.Username))
        {
            optionsBuilder.WithCredentials(config.Username, config.Password);
        }

        var options = optionsBuilder.Build();

        await LogMqttInfoAsync("Connecting to MQTT broker at {Host}:{Port}...", config.Host, config.Port);
        
        try
        {
            await _mqttClient.ConnectAsync(options, _cts.Token);
            await LogMqttInfoAsync("Successfully connected to MQTT broker.");
            
            await ResubscribeAllAsync();
            await OnConnectedAsync();
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to establish initial connection to MQTT broker. Reconnection loop will handle recovery.");
        }
    }

    private bool _disposed;

    /// <summary>
    /// Gracefully disconnects from the broker and stops all ongoing background loops.
    /// </summary>
    public async Task DisconnectAsync()
    {
        if (_disposed) return;
        
        if (_cts != null)
        {
            try
            {
                await _cts.CancelAsync();
            }
            catch (ObjectDisposedException) { }
        }

        if (_mqttClient.IsConnected)
        {
            try
            {
                await _mqttClient.DisconnectAsync(new MqttClientDisconnectOptions 
                { 
                    Reason = MqttClientDisconnectOptionsReason.NormalDisconnection 
                });
                await LogMqttInfoAsync("MQTT connection gracefully closed.");
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "An error occurred while disconnecting from the MQTT broker.");
            }
        }
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter without attaching a direct handler.
    /// </summary>
    public async Task SubscribeAsync(string topic, bool noLocal = false)
    {
        _subscribedTopics[topic] = noLocal;

        if (!IsConnected)
        {
            _logger.LogWarning("Subscription to topic '{Topic}' queued: Client is currently offline.", topic);
            return;
        }

        await ExecuteSubscriptionAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter and maps an asynchronous raw string handler (topic, payload).
    /// </summary>
    public async Task SubscribeAsync(string topic, Func<string, string, Task> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, handler);
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter and maps a synchronous raw string handler (topic, payload).
    /// </summary>
    public async Task SubscribeAsync(string topic, Action<string, string> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, (t, p) => { handler(t, p); return Task.CompletedTask; });
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter and maps an asynchronous handler receiving the raw payload string.
    /// </summary>
    public async Task SubscribeAsync(string topic, Func<string, Task> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, (_, p) => handler(p));
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter, automatically deserializes the JSON payload into <typeparamref name="TPayload"/>,
    /// extracts the correlation ID, and invokes the asynchronous handler with (payload, topic, correlationId).
    /// </summary>
    public async Task SubscribeAsync<TPayload>(string topic, Func<TPayload?, string, string, Task> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, async (t, p) =>
        {
            var data = DeserializePayload<TPayload>(p);
            var correlationId = ExtractCorrelationId(t);
            await handler(data, t, correlationId);
        });
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter, automatically deserializes the JSON payload into <typeparamref name="TPayload"/>,
    /// extracts the correlation ID, and invokes the synchronous handler with (payload, topic, correlationId).
    /// </summary>
    public async Task SubscribeAsync<TPayload>(string topic, Action<TPayload?, string, string> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, (t, p) =>
        {
            var data = DeserializePayload<TPayload>(p);
            var correlationId = ExtractCorrelationId(t);
            handler(data, t, correlationId);
            return Task.CompletedTask;
        });
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Registers a server-side RPC handler that listens on a request topic (e.g. "services/request/assets_Get/#"),
    /// executes the delegate, and publishes the returned <typeparamref name="TResponse"/> to "services/response/{channel}/{correlationId}".
    /// If the request payload cannot be deserialized into <typeparamref name="TRequest"/>, or if
    /// <paramref name="handler"/> throws, no response is silently dropped: a typed <see cref="RpcErrorResponse"/>
    /// fault is published instead (see <see cref="PublishRpcFaultAsync"/>), so a caller using
    /// <see cref="SendRpcRequestAsync{TResponse,TRequest}"/> observes a specific fault instead of only ever
    /// hitting its request timeout.
    /// </summary>
    public async Task SubscribeRpcAsync<TRequest, TResponse>(string requestTopic, Func<TRequest?, string, Task<TResponse>> handler, bool noLocal = false)
    {
        RegisterTopicHandler(requestTopic, async (t, p) =>
        {
            var correlationId = ExtractCorrelationId(t);
            if (string.IsNullOrEmpty(correlationId)) return;

            var segments = t.Split('/', StringSplitOptions.RemoveEmptyEntries);
            var channel = segments.Length >= 3 ? segments[2] : "unknown";
            var responseTopic = MqttTopics.ResponseTopic(channel, correlationId);

            TRequest? req;
            try
            {
                req = DeserializePayload<TRequest>(p);
            }
            catch (Exception ex)
            {
                await PublishRpcFaultAsync(responseTopic, RpcFaultCode.InvalidArgument,
                    "The request payload could not be parsed.", ex);
                return;
            }

            TResponse result;
            try
            {
                result = await handler(req, correlationId);
            }
            catch (Exception ex)
            {
                await PublishRpcFaultAsync(responseTopic, ClassifyFault(ex), SafeFaultMessage(ex), ex);
                return;
            }

            await PublishAsync(responseTopic, result);
        });
        await SubscribeAsync(requestTopic, noLocal);
    }

    /// <summary>
    /// Maps an exception thrown by an RPC handler onto the small, coarse <see cref="RpcFaultCode"/> set so the
    /// caller-side <see cref="SendRpcRequestAsync{TResponse,TRequest}"/> can reconstruct an equivalent standard
    /// .NET exception type across the MQTT boundary (see <see cref="RpcFaultCode"/> for the mapping rationale).
    /// </summary>
    /// <param name="ex">The exception thrown by the RPC handler.</param>
    /// <returns>The fault classification to report to the caller.</returns>
    private static RpcFaultCode ClassifyFault(Exception ex) => ex switch
    {
        ArgumentException => RpcFaultCode.InvalidArgument,
        KeyNotFoundException => RpcFaultCode.NotFound,
        UnauthorizedAccessException => RpcFaultCode.Unauthorized,
        InvalidOperationException => RpcFaultCode.Conflict,
        _ => RpcFaultCode.Internal
    };

    /// <summary>
    /// Produces the message text that is safe to place on the (currently unauthenticated) MQTT broker for a
    /// given RPC handler exception. Exceptions that already carry a deliberately-authored, business-facing
    /// message (the four types <see cref="ClassifyFault"/> recognizes) are passed through as-is; anything else
    /// is replaced with a generic message, since it may be an unexpected infrastructure failure whose message
    /// could contain internal details. The original exception (including its stack trace) is always logged
    /// locally by <see cref="PublishRpcFaultAsync"/> regardless of which branch is taken.
    /// </summary>
    /// <param name="ex">The exception thrown by the RPC handler.</param>
    /// <returns>A short, safe message describing the fault to an external caller.</returns>
    private static string SafeFaultMessage(Exception ex) => ex switch
    {
        ArgumentException or KeyNotFoundException or UnauthorizedAccessException or InvalidOperationException
            => ex.Message,
        _ => "An internal error occurred while processing the request."
    };

    /// <summary>
    /// Logs an RPC handler fault locally (with full exception detail) and publishes a corresponding
    /// <see cref="RpcErrorResponse"/> to the fault sibling of <paramref name="responseTopic"/> (see
    /// <see cref="ErrorTopicSuffix"/>), so the caller of <see cref="SendRpcRequestAsync{TResponse,TRequest}"/>
    /// observes a typed fault instead of silently timing out. If the fault publish itself fails (e.g. the
    /// broker connection dropped between receiving the request and reporting the fault), that secondary failure
    /// is logged but not rethrown, since the caller's request timeout is still a safe fallback in that case.
    /// </summary>
    /// <param name="responseTopic">The normal ("success") response topic for the failed request.</param>
    /// <param name="code">The machine-readable fault classification to report.</param>
    /// <param name="message">The safe, non-sensitive message to report.</param>
    /// <param name="ex">The original exception, logged locally in full but never placed on the wire.</param>
    private async Task PublishRpcFaultAsync(string responseTopic, RpcFaultCode code, string message, Exception ex)
    {
        _logger.LogError(ex, "RPC handler faulted for response topic '{ResponseTopic}'. Reporting fault {FaultCode} to the caller.", responseTopic, code);

        try
        {
            await PublishAsync(responseTopic + ErrorTopicSuffix, new RpcErrorResponse(code, message));
        }
        catch (Exception publishEx)
        {
            _logger.LogError(publishEx, "Failed to publish RPC fault response to '{ResponseTopic}'; the caller will fall back to its request timeout.", responseTopic + ErrorTopicSuffix);
        }
    }

    /// <summary>
    /// Registers a server-side RPC handler without correlation ID parameter in the delegate.
    /// </summary>
    public async Task SubscribeRpcAsync<TRequest, TResponse>(string requestTopic, Func<TRequest?, Task<TResponse>> handler, bool noLocal = false)
    {
        await SubscribeRpcAsync<TRequest, TResponse>(requestTopic, (req, _) => handler(req), noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter, automatically deserializes the JSON payload into <typeparamref name="TPayload"/>,
    /// and invokes the asynchronous handler with (payload, topic).
    /// </summary>
    public async Task SubscribeAsync<TPayload>(string topic, Func<TPayload?, string, Task> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, async (t, p) =>
        {
            var data = DeserializePayload<TPayload>(p);
            await handler(data, t);
        });
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter, automatically deserializes the JSON payload into <typeparamref name="TPayload"/>,
    /// and invokes the asynchronous handler with the payload.
    /// </summary>
    public async Task SubscribeAsync<TPayload>(string topic, Func<TPayload?, Task> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, async (_, p) =>
        {
            var data = DeserializePayload<TPayload>(p);
            await handler(data);
        });
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter, automatically deserializes the JSON payload into <typeparamref name="TPayload"/>,
    /// and invokes the synchronous handler with (payload, topic).
    /// </summary>
    public async Task SubscribeAsync<TPayload>(string topic, Action<TPayload?, string> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, (t, p) =>
        {
            var data = DeserializePayload<TPayload>(p);
            handler(data, t);
            return Task.CompletedTask;
        });
        await SubscribeAsync(topic, noLocal);
    }

    /// <summary>
    /// Subscribes to a specific MQTT topic filter, automatically deserializes the JSON payload into <typeparamref name="TPayload"/>,
    /// and invokes the synchronous handler with the payload.
    /// </summary>
    public async Task SubscribeAsync<TPayload>(string topic, Action<TPayload?> handler, bool noLocal = false)
    {
        RegisterTopicHandler(topic, (_, p) =>
        {
            var data = DeserializePayload<TPayload>(p);
            handler(data);
            return Task.CompletedTask;
        });
        await SubscribeAsync(topic, noLocal);
    }

    private void RegisterTopicHandler(string topic, Func<string, string, Task> handler)
    {
        _topicHandlers.AddOrUpdate(
            topic,
            _ => new List<Func<string, string, Task>> { handler },
            (_, list) =>
            {
                lock (list)
                {
                    list.Add(handler);
                }
                return list;
            });
    }

    private async Task ExecuteSubscriptionAsync(string topic, bool noLocal)
    {
        var filterBuilder = new MqttTopicFilterBuilder().WithTopic(topic);
        if (noLocal)
        {
            filterBuilder.WithNoLocal();
        }

        var subscribeOptions = new MqttClientFactory().CreateSubscribeOptionsBuilder()
            .WithTopicFilter(filterBuilder.Build())
            .Build();

        await _mqttClient.SubscribeAsync(subscribeOptions, CancellationToken.None);
        await LogMqttDebugAsync("Successfully subscribed to topic: {Topic} (NoLocal: {NoLocal})", topic, noLocal);
    }

    private async Task ResubscribeAllAsync()
    {
        foreach (var kvp in _subscribedTopics)
        {
            try
            {
                await ExecuteSubscriptionAsync(kvp.Key, kvp.Value);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "Failed to re-subscribe to topic '{Topic}' after reconnect.", kvp.Key);
            }
        }
    }

    /// <summary>
    /// Publishes a raw string message payload to the specified topic.
    /// </summary>
    public async Task PublishAsync(string topic, string payload, bool retain = false)
    {
        if (!IsConnected)
            throw new InvalidOperationException("Cannot publish message: MQTT client is offline.");

        var message = new MqttApplicationMessageBuilder()
            .WithTopic(topic)
            .WithPayload(payload)
            .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
            .WithRetainFlag(retain)
            .Build();

        await _mqttClient.PublishAsync(message, CancellationToken.None);
    }

    /// <summary>
    /// Serializes a generic object into a structured JSON string and publishes it to the specified topic.
    /// Uses standard System.Text.Json with fallback to Source Generators.
    /// </summary>
    public Task PublishAsync<T>(string topic, T data, bool retain = false)
    {
        if (!IsConnected)
            throw new InvalidOperationException("Cannot publish message: MQTT client is offline.");

        byte[] jsonBytes;
        if (data is string str)
        {
            jsonBytes = Encoding.UTF8.GetBytes(str);
        }
        else if (data is byte[] b)
        {
            jsonBytes = b;
        }
        else
        {
            try
            {
                jsonBytes = JsonSerializer.SerializeToUtf8Bytes(data, DefaultJsonOptions);
            }
            catch
            {
                var typeInfo = FinlyticJsonSerializerContext.Default.GetTypeInfo(typeof(T))
                               ?? (data != null ? FinlyticJsonSerializerContext.Default.GetTypeInfo(data.GetType()) : null);

                if (typeInfo != null)
                {
                    jsonBytes = JsonSerializer.SerializeToUtf8Bytes(data, typeInfo);
                }
                else
                {
                    jsonBytes = JsonSerializer.SerializeToUtf8Bytes(data);
                }
            }
        }

        var message = new MqttApplicationMessageBuilder()
            .WithTopic(topic)
            .WithPayload(jsonBytes)
            .WithQualityOfServiceLevel(MQTTnet.Protocol.MqttQualityOfServiceLevel.AtLeastOnce)
            .WithRetainFlag(retain)
            .Build();

        return _mqttClient.PublishAsync(message, CancellationToken.None);
    }

    /// <summary>
    /// Sends a parameterless request to an RPC channel and asynchronously blocks until a matching response arrives.
    /// See <see cref="SendRpcRequestAsync{TResponse,TRequest}"/> for the exact timeout/fault-propagation contract.
    /// </summary>
    public Task<TResponse?> SendRpcRequestAsync<TResponse>(
        string channel, 
        TimeSpan? timeout = null) 
        where TResponse : class
    {
        return SendRpcRequestAsync<TResponse, string>(channel, string.Empty, timeout);
    }

    /// <summary>
    /// Sends a generic request payload to an RPC channel and asynchronously waits for a matching response.
    /// See <see cref="SendRpcRequestAsync{TResponse,TRequest}"/> for the exact timeout/fault-propagation contract.
    /// </summary>
    public Task<TResponse?> RequestAsync<TRequest, TResponse>(
        string channel, 
        TRequest requestData, 
        TimeSpan? timeout = null) 
        where TResponse : class 
        where TRequest : class
    {
        var cleanChannel = channel.StartsWith(MqttTopics.RequestPrefix) ? channel.Substring(MqttTopics.RequestPrefix.Length).TrimEnd('/') : channel;
        return SendRpcRequestAsync<TResponse, TRequest>(cleanChannel, requestData, timeout);
    }

    /// <summary>
    /// Sends a generic request payload to an RPC channel and asynchronously blocks until a matching response arrives.
    /// Uses the topic conventions: <c>services/request/{channel}/{correlationId}</c> and <c>services/response/{channel}/{correlationId}</c>.
    /// If the serving handler faulted, the server publishes an <see cref="RpcErrorResponse"/> on the sibling
    /// error topic (<see cref="ErrorTopicSuffix"/>) instead of the normal response; this method then throws a
    /// reconstructed exception (an <see cref="ArgumentException"/>, <see cref="InvalidOperationException"/>,
    /// <see cref="KeyNotFoundException"/>, <see cref="UnauthorizedAccessException"/>, or, for anything that does
    /// not map onto one of those, an <see cref="RpcFaultException"/>) instead of returning. This lets a caller
    /// distinguish a specific server-side fault from an unreachable/silent server, which still surfaces as a
    /// <see cref="TimeoutException"/>-driven <c>null</c> return exactly as before this fault channel existed.
    /// </summary>
    /// <exception cref="ArgumentException">The remote handler reported <see cref="RpcFaultCode.InvalidArgument"/>.</exception>
    /// <exception cref="InvalidOperationException">The remote handler reported <see cref="RpcFaultCode.Conflict"/>, or the client is offline.</exception>
    /// <exception cref="KeyNotFoundException">The remote handler reported <see cref="RpcFaultCode.NotFound"/>.</exception>
    /// <exception cref="UnauthorizedAccessException">The remote handler reported <see cref="RpcFaultCode.Unauthorized"/>.</exception>
    /// <exception cref="RpcFaultException">The remote handler reported <see cref="RpcFaultCode.Internal"/>, or its fault payload could not be parsed.</exception>
    public async Task<TResponse?> SendRpcRequestAsync<TResponse, TRequest>(
        string channel, 
        TRequest requestData, 
        TimeSpan? timeout = null) 
        where TResponse : class 
        where TRequest : class
    {
        if (!IsConnected)
            throw new InvalidOperationException("Cannot execute RPC request: MQTT client is offline.");

        string correlationId = Guid.NewGuid().ToString("N");
        
        var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        _pendingRequests.TryAdd(correlationId, tcs);

        string requestTopic = MqttTopics.RequestTopic(channel, correlationId);

        await PublishAsync(requestTopic, requestData);
        await LogMqttInfoAsync("RPC request published to '{Topic}' [CorrelationId: {Id}]", requestTopic, correlationId);

        try
        {
            var effectiveTimeout = timeout ?? TimeSpan.FromSeconds(25);
            var rawJsonResult = await tcs.Task.WaitAsync(effectiveTimeout);

            if (typeof(TResponse) == typeof(string))
            {
                return rawJsonResult as TResponse;
            }

            return DeserializePayload<TResponse>(rawJsonResult);
        }
        catch (TimeoutException)
        {
            _logger.LogWarning("RPC request timed out on channel '{Channel}' [CorrelationId: {Id}]", channel, correlationId);
            return null;
        }
        finally
        {
            _pendingRequests.TryRemove(correlationId, out _);
        }
    }

    private Task HandleIncomingMessageAsync(MqttApplicationMessageReceivedEventArgs e)
    {
        _ = Task.Run(async () =>
        {
            try
            {
                var topic = e.ApplicationMessage.Topic;
                var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
                await LogMqttDebugAsync("MQTT message received on topic '{Topic}', length={Length}", topic, payload?.Length ?? 0);

                // Intercept message if it belongs to the RPC response convention
                if (topic.StartsWith(MqttTopics.ResponsePrefix))
                {
                    // A fault sibling topic ends in ErrorTopicSuffix (see SubscribeRpcAsync/PublishRpcFaultAsync);
                    // strip it before extracting the correlation ID so both topic shapes resolve the same pending
                    // request. An old client build (pre-dating this suffix) would instead extract "error" itself
                    // as a bogus correlation ID, find no matching pending request below, and fall through to time
                    // out exactly as it did before this fault channel existed - see ErrorTopicSuffix remarks.
                    bool isFault = topic.EndsWith(ErrorTopicSuffix, StringComparison.Ordinal);
                    var correlationTopic = isFault ? topic[..^ErrorTopicSuffix.Length] : topic;

                    var lastSlashIndex = correlationTopic.LastIndexOf('/');
                    if (lastSlashIndex != -1)
                    {
                        string correlationId = correlationTopic[(lastSlashIndex + 1)..];

                        if (_pendingRequests.TryRemove(correlationId, out var tcs))
                        {
                            if (isFault)
                            {
                                tcs.SetException(BuildFaultException(payload ?? string.Empty));
                            }
                            else
                            {
                                tcs.SetResult(payload ?? string.Empty);
                            }
                            return; // Sinks the message, avoiding triggering OnMessageReceivedAsync for active RPC handles
                        }
                    }
                }

                // Match registered topic handlers
                foreach (var kvp in _topicHandlers)
                {
                    if (TopicMatches(kvp.Key, topic))
                    {
                        List<Func<string, string, Task>> handlersCopy;
                        lock (kvp.Value)
                        {
                            handlersCopy = new List<Func<string, string, Task>>(kvp.Value);
                        }

                        for (int i = 0; i < handlersCopy.Count; i++)
                        {
                            try
                            {
                                await handlersCopy[i](topic, payload ?? string.Empty);
                            }
                            catch (Exception ex)
                            {
                                OnError(ex);
                            }
                        }
                    }
                }

                // Regular Pub/Sub message propagation (for overridden OnMessageReceivedAsync)
                await OnMessageReceivedAsync(topic, payload ?? string.Empty);
            }
            catch (Exception ex)
            {
                OnError(ex);
            }
        });

        return Task.CompletedTask;
    }

    private async Task HandleDisconnectAsync(MqttClientDisconnectedEventArgs e)
    {
        if (_cts == null || _cts.IsCancellationRequested) 
            return; 

        _logger.LogWarning("Lost connection to MQTT broker (Reason: {Reason}). Initiating auto-reconnect loop...", e.Reason);

        int attempt = 0;
        while (_cts != null && !_cts.IsCancellationRequested)
        {
            attempt++;
            var delaySeconds = Math.Min(5 * Math.Pow(2, attempt - 1), 60); // 5s, 10s, 20s, 40s, 60s max

            try
            {
                await LogMqttInfoAsync("Reconnect attempt {Attempt} in {Delay}s...", attempt, delaySeconds);
                await Task.Delay(TimeSpan.FromSeconds(delaySeconds), _cts.Token);
                
                await _mqttClient.ReconnectAsync(_cts.Token);
                
                if (_mqttClient.IsConnected)
                {
                    await LogMqttInfoAsync("MQTT client reconnected successfully after {Attempt} attempt(s).", attempt);
                    await ResubscribeAllAsync();
                    await OnConnectedAsync();
                    return;
                }
            }
            catch (OperationCanceledException) { return; }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "Reconnect attempt {Attempt} to the MQTT broker failed.", attempt);
            }
        }
    }

    /// <summary>
    /// Deserializes a JSON string payload into <typeparamref name="T"/> using standard System.Text.Json with fallback.
    /// </summary>
    public static T? DeserializePayload<T>(string payload)
    {
        if (string.IsNullOrWhiteSpace(payload)) return default;
        if (typeof(T) == typeof(string)) return (T)(object)payload;

        try
        {
            return JsonSerializer.Deserialize<T>(payload, DefaultJsonOptions);
        }
        catch
        {
            var typeInfo = FinlyticJsonSerializerContext.Default.GetTypeInfo(typeof(T));
            if (typeInfo != null)
            {
                return (T?)JsonSerializer.Deserialize(payload, typeInfo);
            }
            throw;
        }
    }

    /// <summary>
    /// Reconstructs the exception a caller should observe for a fault reported on an RPC error topic (see
    /// <see cref="ErrorTopicSuffix"/> / <see cref="PublishRpcFaultAsync"/>). Faults whose
    /// <see cref="RpcErrorResponse.Code"/> maps onto a familiar .NET exception type are thrown as that type
    /// (see <see cref="RpcFaultCode"/>), so pre-existing <c>catch</c> blocks written against the underlying
    /// service-layer exception types (e.g. in <c>FinlyticBackend</c> controllers) start working across the MQTT
    /// boundary without any changes on the caller's side. Anything else, including a fault payload that fails
    /// to parse, becomes an <see cref="RpcFaultException"/>.
    /// </summary>
    /// <param name="payload">The raw JSON payload received on the fault topic.</param>
    /// <returns>The exception to throw to the RPC caller.</returns>
    private Exception BuildFaultException(string payload)
    {
        RpcErrorResponse? fault;
        try
        {
            fault = DeserializePayload<RpcErrorResponse>(payload);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to parse RPC fault payload; propagating a generic RpcFaultException instead.");
            return new RpcFaultException(RpcFaultCode.Internal, "The remote service reported an error that could not be parsed.");
        }

        if (fault == null)
        {
            return new RpcFaultException(RpcFaultCode.Internal, "The remote service reported an empty error response.");
        }

        return fault.Code switch
        {
            RpcFaultCode.InvalidArgument => new ArgumentException(fault.Message),
            RpcFaultCode.Conflict => new InvalidOperationException(fault.Message),
            RpcFaultCode.NotFound => new KeyNotFoundException(fault.Message),
            RpcFaultCode.Unauthorized => new UnauthorizedAccessException(fault.Message),
            _ => new RpcFaultException(fault.Code, fault.Message)
        };
    }

    /// <summary>
    /// Checks whether an MQTT topic matches a topic filter with wildcards ('+' and '#').
    /// </summary>
    public static bool TopicMatches(string filter, string topic)
    {
        if (string.Equals(filter, topic, StringComparison.OrdinalIgnoreCase)) return true;
        if (filter == "#") return true;

        var filterSegments = filter.Split('/');
        var topicSegments = topic.Split('/');

        for (int i = 0; i < filterSegments.Length; i++)
        {
            var f = filterSegments[i];
            if (f == "#")
            {
                return true;
            }

            if (i >= topicSegments.Length)
            {
                return false;
            }

            var t = topicSegments[i];
            if (f != "+" && !string.Equals(f, t, StringComparison.OrdinalIgnoreCase))
            {
                return false;
            }
        }

        return filterSegments.Length == topicSegments.Length;
    }

    /// <summary>
    /// Extracts the Correlation ID from the end of an RPC request or response topic (e.g. services/request/abc/123 -> 123).
    /// </summary>
    public static string ExtractCorrelationId(string topic)
    {
        if (string.IsNullOrWhiteSpace(topic)) return string.Empty;
        var lastSlash = topic.LastIndexOf('/');
        return lastSlash >= 0 && lastSlash < topic.Length - 1 
            ? topic[(lastSlash + 1)..] 
            : string.Empty;
    }

    /// <summary>
    /// Fired automatically whenever a connection or reconnection is successfully established.
    /// </summary>
    protected abstract Task OnConnectedAsync();

    /// <summary>
    /// Fired whenever a new message lands on a registered subscription channel.
    /// </summary>
    protected virtual Task OnMessageReceivedAsync(string topic, string payload) => Task.CompletedTask;

    /// <summary>
    /// Virtual fallback method to catch and handle processing level exceptions inside the incoming pipeline.
    /// </summary>
    protected virtual void OnError(Exception ex)
    {
        _logger.LogError(ex, "An unhandled exception occurred within the ManagedMqttClient messaging pipeline.");
    }

    public void Dispose()
    {
        if (_disposed) return;
        _disposed = true;
        
        try { DisconnectAsync().GetAwaiter().GetResult(); } catch { }
        _cts?.Dispose();
        _mqttClient.Dispose();
        GC.SuppressFinalize(this);
    }
}

/// <summary>
/// Thrown client-side by <see cref="ManagedMqttClient.SendRpcRequestAsync{TResponse,TRequest}"/> when a remote
/// RPC handler reported a fault (<see cref="RpcErrorResponse"/>) whose <see cref="RpcFaultCode"/> has no
/// equivalent standard .NET exception type — i.e. <see cref="Dtos.RpcFaultCode.Internal"/>, or a fault payload
/// that could not be parsed at all. Faults that DO map onto an existing exception type
/// (<see cref="Dtos.RpcFaultCode.InvalidArgument"/> to <see cref="ArgumentException"/>,
/// <see cref="Dtos.RpcFaultCode.Conflict"/> to <see cref="InvalidOperationException"/>,
/// <see cref="Dtos.RpcFaultCode.NotFound"/> to <see cref="KeyNotFoundException"/>,
/// <see cref="Dtos.RpcFaultCode.Unauthorized"/> to <see cref="UnauthorizedAccessException"/>) are deliberately
/// thrown as that familiar type instead of this one: several existing callers (e.g.
/// <c>FinlyticBackend/Controllers/UserTradesController.cs</c>) already have <c>catch (InvalidOperationException)</c>
/// / <c>catch (ArgumentException)</c> blocks written for the exception types the underlying service-layer
/// methods throw locally, and reusing those types here reactivates that existing code instead of requiring
/// every caller to learn and catch a brand new exception type.
/// </summary>
public sealed class RpcFaultException : Exception
{
    /// <summary>Gets the machine-readable fault classification reported by the remote RPC handler.</summary>
    public RpcFaultCode Code { get; }

    /// <summary>
    /// Initializes a new instance carrying the remote fault's classification and its safe, non-sensitive message.
    /// </summary>
    /// <param name="code">The machine-readable fault classification reported by the remote RPC handler.</param>
    /// <param name="message">The safe, non-sensitive message reported by the remote handler.</param>
    public RpcFaultException(RpcFaultCode code, string message) : base(message)
    {
        Code = code;
    }
}