feat(core): add shared DTOs, MqttTopics constants, DatabaseBootstrapper, and ManagedMqttClient extensions

2026-08-24 21:35:24 +02:00
parent 6ab84fe1de
commit 44b161d509
39 changed files with 2545 additions and 709 deletions
@@ -1,10 +1,12 @@
 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;
@@ -15,11 +17,19 @@ namespace FinlyticCore.Util;

 /// <summary>
 /// An abstract, resilient MQTT client wrapper designed for microservice architectures.
-/// Handles automatic reconnection, structured JSON publishing, thread-safe subscription management, and synchronous Request-Reply (RPC).
+/// 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;
@@ -29,6 +39,27 @@ public abstract class ManagedMqttClient : IDisposable
    // 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>
@@ -89,7 +120,6 @@ public abstract class ManagedMqttClient : IDisposable
    /// <summary>
    /// Establishes a connection to the MQTT broker and initializes the background auto-reconnection loop.
    /// </summary>
-    /// <param name="config">The network and credential configuration options for the broker.</param>
    public async Task ConnectAsync(MqttConfiguration config)
    {
        if (IsConnected)
@@ -116,6 +146,7 @@ public abstract class ManagedMqttClient : IDisposable
            await _mqttClient.ConnectAsync(options, _cts.Token);
            await LogMqttInfoAsync("Successfully connected to MQTT broker.");
            
+            await ResubscribeAllAsync();
            await OnConnectedAsync();
        }
        catch (Exception ex)
@@ -124,14 +155,22 @@ public abstract class ManagedMqttClient : IDisposable
        }
    }

+    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)
        {
-            await _cts.CancelAsync();
+            try
+            {
+                await _cts.CancelAsync();
+            }
+            catch (ObjectDisposedException) { }
        }

        if (_mqttClient.IsConnected)
@@ -152,18 +191,269 @@ public abstract class ManagedMqttClient : IDisposable
    }

    /// <summary>
-    /// Subscribes to a specific MQTT topic filter.
+    /// Subscribes to a specific MQTT topic filter without attaching a direct handler.
    /// </summary>
-    /// <param name="topic">The topic pattern or wildcard to subscribe to.</param>
-    /// <param name="noLocal">If set to <c>true</c>, the broker will not forward messages published by this client back to itself.</param>
-    protected async Task SubscribeAsync(string topic, bool noLocal = false)
+    public async Task SubscribeAsync(string topic, bool noLocal = false)
    {
+        _subscribedTopics[topic] = noLocal;
+
        if (!IsConnected)
        {
-            _logger.LogWarning("Subscription to topic '{Topic}' delayed: Client is currently offline.", topic);
+            _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)
        {
@@ -178,6 +468,21 @@ public abstract class ManagedMqttClient : IDisposable
        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>
@@ -198,25 +503,43 @@ public abstract class ManagedMqttClient : IDisposable

    /// <summary>
    /// Serializes a generic object into a structured JSON string and publishes it to the specified topic.
-    /// Utilizes .NET 8 JSON Source Generators for zero-reflection overhead, with reflection fallback for unregistered types.
+    /// Uses standard System.Text.Json with fallback to Source Generators.
    /// </summary>
    public Task PublishAsync<T>(string topic, T data, bool retain = false)
    {
-        byte[] jsonBytes;
-        var typeInfo = FinlyticJsonSerializerContext.Default.GetTypeInfo(typeof(T))
-                       ?? (data != null ? FinlyticJsonSerializerContext.Default.GetTypeInfo(data.GetType()) : null);
+        if (!IsConnected)
+            throw new InvalidOperationException("Cannot publish message: MQTT client is offline.");

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

-        if (!IsConnected)
-            throw new InvalidOperationException("Cannot publish message: MQTT client is offline.");
+                if (typeInfo != null)
+                {
+                    jsonBytes = JsonSerializer.SerializeToUtf8Bytes(data, typeInfo);
+                }
+                else
+                {
+                    jsonBytes = JsonSerializer.SerializeToUtf8Bytes(data);
+                }
+            }
+        }

        var message = new MqttApplicationMessageBuilder()
            .WithTopic(topic)
@@ -230,6 +553,7 @@ public abstract class ManagedMqttClient : IDisposable

    /// <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, 
@@ -239,10 +563,37 @@ public abstract class ManagedMqttClient : IDisposable
        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, 
@@ -253,21 +604,18 @@ public abstract class ManagedMqttClient : IDisposable
        if (!IsConnected)
            throw new InvalidOperationException("Cannot execute RPC request: MQTT client is offline.");

-        // 1. Generate a unique Correlation ID for this specific transaction
        string correlationId = Guid.NewGuid().ToString("N");
        
        var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
        _pendingRequests.TryAdd(correlationId, tcs);

-        string requestTopic = $"services/request/{channel}/{correlationId}";
+        string requestTopic = MqttTopics.RequestTopic(channel, correlationId);

-        // 2. Serialize and dispatch via the existing JSON helper
        await PublishAsync(requestTopic, requestData);
        await LogMqttInfoAsync("RPC request published to '{Topic}' [CorrelationId: {Id}]", requestTopic, correlationId);

        try
        {
-            // 3. Block asynchronously until the response loop resolves the token
            var effectiveTimeout = timeout ?? TimeSpan.FromSeconds(25);
            var rawJsonResult = await tcs.Task.WaitAsync(effectiveTimeout);

@@ -276,13 +624,7 @@ public abstract class ManagedMqttClient : IDisposable
                return rawJsonResult as TResponse;
            }

-            var respTypeInfo = FinlyticJsonSerializerContext.Default.GetTypeInfo(typeof(TResponse));
-            if (respTypeInfo != null)
-            {
-                return JsonSerializer.Deserialize(rawJsonResult, respTypeInfo) as TResponse;
-            }
-
-            return JsonSerializer.Deserialize<TResponse>(rawJsonResult);
+            return DeserializePayload<TResponse>(rawJsonResult);
        }
        catch (TimeoutException)
        {
@@ -291,7 +633,6 @@ public abstract class ManagedMqttClient : IDisposable
        }
        finally
        {
-            // Always clean up the dictionary to prevent memory leaks
            _pendingRequests.TryRemove(correlationId, out _);
        }
    }
@@ -307,22 +648,62 @@ public abstract class ManagedMqttClient : IDisposable
                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("services/response/"))
+                if (topic.StartsWith(MqttTopics.ResponsePrefix))
                {
-                    var lastSlashIndex = topic.LastIndexOf('/');
+                    // 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 = topic[(lastSlashIndex + 1)..];
+                        string correlationId = correlationTopic[(lastSlashIndex + 1)..];

                        if (_pendingRequests.TryRemove(correlationId, out var tcs))
                        {
-                            tcs.SetResult(payload ?? string.Empty);
+                            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
                        }
                    }
                }

-                // Regular Pub/Sub message propagation
+                // 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)
@@ -357,6 +738,7 @@ public abstract class ManagedMqttClient : IDisposable
                if (_mqttClient.IsConnected)
                {
                    await LogMqttInfoAsync("MQTT client reconnected successfully after {Attempt} attempt(s).", attempt);
+                    await ResubscribeAllAsync();
                    await OnConnectedAsync();
                    return;
                }
@@ -369,6 +751,114 @@ public abstract class ManagedMqttClient : IDisposable
        }
    }

+    /// <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>
@@ -377,7 +867,7 @@ public abstract class ManagedMqttClient : IDisposable
    /// <summary>
    /// Fired whenever a new message lands on a registered subscription channel.
    /// </summary>
-    protected abstract Task OnMessageReceivedAsync(string topic, string payload);
+    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.
@@ -389,9 +879,43 @@ public abstract class ManagedMqttClient : IDisposable

    public void Dispose()
    {
-        DisconnectAsync().GetAwaiter().GetResult();
+        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;
+    }
 }