Kleidukos/Finlytic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(Assets): update asset services and background workers

Browse Source
This commit is contained in:
Kleidukos
2026-08-09 21:01:39 +02:00
parent b57cc9894c
commit e0778b88ea
32 changed files with 1020 additions and 2062 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticAssets/Util/AssetsMqttClient.cs
+162 -36
View File
@@ -1,46 +1,74 @@
using System.Text.Json;
using System.Text.Json;
using FinlyticAssets.Services;
using FinlyticCore.Entities.Assets;
using FinlyticCore.Models;
using FinlyticCore.Models.Assets;
using FinlyticCore.Util;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace FinlyticAssets.Util;
/// <summary>
/// Represents a managed MQTT client acting as a server-side RPC provider within the asset microservice.
/// It subscribes to request topics, processes incoming JSON payloads via the database service, and publishes
/// the requested asset entities back to the corresponding response topic.
/// It subscribes to request topics, processes incoming JSON payloads via the database and index service,
/// and publishes the requested asset entities or logo files back to the response topic.
/// Also implements <see cref="IHostedService"/> to manage its own lifecycle connections.
/// </summary>
public class AssetsMqttClient : ManagedMqttClient
public class AssetsMqttClient(
ILogger<AssetsMqttClient> logger,
IServiceScopeFactory scopeFactory,
IConfiguration configuration) : ManagedMqttClient(logger), IHostedService
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger<AssetsMqttClient> _logger;
/// <summary>
/// Initializes a new instance of the <see cref="AssetsMqttClient"/> class.
/// Starts the MQTT client and connects to the configured broker.
/// </summary>
/// <param name="logger">The logger used to record connection, error, and status messages.</param>
/// <param name="dbService">The database service used for querying and validating assets.</param>
public AssetsMqttClient(ILogger<AssetsMqttClient> logger, IServiceScopeFactory scopeFactory)
: base(logger)
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A task representing the asynchronous start operation.</returns>
public async Task StartAsync(CancellationToken cancellationToken)
{
_scopeFactory = scopeFactory;
_logger = logger;
var config = new MqttConfiguration()
{
Host = configuration["MQTT:Host"] ?? configuration["MQTT__Host"]!,
Port = Convert.ToInt32(configuration["MQTT:Port"] ?? configuration["MQTT__Port"]!),
ClientId = $"{(configuration["MQTT:ClientId"] ?? configuration["MQTT__ClientId"]!)}_{Guid.NewGuid()}"
};
await ConnectAsync(config);
}
/// <summary>
/// Gracefully stops and disconnects the MQTT client.
/// </summary>
/// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
/// <returns>A task representing the asynchronous stop operation.</returns>
public async Task StopAsync(CancellationToken cancellationToken)
{
await DisconnectAsync();
}
/// <summary>
/// Invoked automatically once the connection to the MQTT broker is successfully established or restored.
/// Registers the required wildcard subscriptions for incoming asset validation and search requests.
/// Registers subscriptions for asset validation, search, and logo download requests.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous subscription operation.</returns>
protected override async Task OnConnectedAsync()
{
await SubscribeAsync("services/request/assets_Get/#");
await SubscribeAsync("services/request/assets_Search/#");
await SubscribeAsync("services/request/assets_GetDiscovery/#");
await SubscribeAsync("services/request/assets_FetchLogo/#");
await SubscribeAsync("services/request/health_Ping/#");
await SubscribeAsync("services/config/updated/#");
}
/// <summary>
/// Processes incoming messages on the subscribed topics, executes the corresponding database query,
/// Processes incoming messages on the subscribed topics, executes the corresponding service methods,
/// and publishes the result to the response topic while preserving the correlation ID.
/// </summary>
/// <param name="topic">The MQTT topic on which the message was received.</param>
@@ -48,46 +76,144 @@ public class AssetsMqttClient : ManagedMqttClient
/// <returns>A <see cref="Task"/> representing the asynchronous message processing operation.</returns>
protected override async Task OnMessageReceivedAsync(string topic, string payload)
{
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
{
await HandleConfigUpdatedAsync(topic, payload);
return;
}
var segments = topic.Split('/');
if (segments.Length < 4) return;
var channel = segments[2];
var correlationId = segments[3];
var correlationId = segments[segments.Length - 1];
if (topic.Contains("health_Ping", StringComparison.OrdinalIgnoreCase))
{
await HandleHealthPingAsync(topic, segments, correlationId);
return;
}
try
{
List<AssetEntity> responseData = [];
using var scope = _scopeFactory.CreateScope();
using var scope = scopeFactory.CreateScope();
var dbService = scope.ServiceProvider.GetRequiredService<IAssetsDbService>();
var indexService = scope.ServiceProvider.GetRequiredService<IAssetsIndexService>();
if (channel == "assets_FetchLogo")
{
await HandleFetchLogoAsync(payload, correlationId, indexService);
return;
}
List<AssetEntity> responseData = [];
switch (channel)
{
case "assets_Get":
var validReq = JsonSerializer.Deserialize<GetValidAssetRequest>(payload);
if (validReq != null)
{
responseData = await dbService.GetValidAssetsByIsinAsync(validReq.Isin);
}
responseData = await HandleAssetsGetAsync(payload, dbService);
break;
case "assets_Search":
var searchReq = JsonSerializer.Deserialize<SearchAssetsRequest>(payload);
if (searchReq != null)
{
responseData = await dbService.FindAffectedActiveAssetsAsync(searchReq.SearchQuery);
}
responseData = await HandleAssetsSearchAsync(payload, dbService);
break;
case "assets_GetDiscovery":
responseData = await HandleAssetsGetDiscoveryAsync(payload, dbService);
break;
}
{
string responseTopic = $"services/response/{channel}/{correlationId}";
await PublishAsync(responseTopic, responseData.ToDtoList());
}
string defaultResponseTopic = $"services/response/{channel}/{correlationId}";
await PublishAsync(defaultResponseTopic, responseData.ToDtoList());
}
catch (Exception ex)
{
OnError(ex);
}
}
}
private async Task HandleConfigUpdatedAsync(string topic, string payload)
{
if (!topic.EndsWith("FinlyticAssets", StringComparison.OrdinalIgnoreCase))
return;
logger.LogInformation("[{Channel}] [AssetsMqttClient] Received config update event for FinlyticAssets.", "AssetsChannel");
try
{
var updatePayload = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
if (updatePayload?.Settings != null && updatePayload.Settings.Count > 0)
{
using var scope = scopeFactory.CreateScope();
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
await settingsDb.UpdateSettingsFromDictionary(updatePayload.Settings);
logger.LogInformation("[{Channel}] [AssetsMqttClient] Persisted {Count} updated settings to FinlyticAssets database.", "AssetsChannel", updatePayload.Settings.Count);
}
}
catch (Exception ex)
{
logger.LogError(ex, "[{Channel}] [AssetsMqttClient] Error processing MQTT config update event.", "AssetsChannel");
}
}
private async Task HandleHealthPingAsync(string topic, string[] segments, string correlationId)
{
bool isForMe = segments.Length >= 5
? segments[3].Equals("FinlyticAssets", StringComparison.OrdinalIgnoreCase)
: topic.Contains("FinlyticAssets", StringComparison.OrdinalIgnoreCase);
if (isForMe)
{
string respTopic = $"services/response/health_Ping/{correlationId}";
await PublishAsync(respTopic, new FinlyticCore.Dtos.ServiceHealthResponse("FinlyticAssets", "Online", DateTime.UtcNow, "Connected"));
logger.LogInformation("[{Channel}] [AssetsMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "AssetsChannel", correlationId);
}
}
private async Task HandleFetchLogoAsync(string payload, string correlationId, IAssetsIndexService indexService)
{
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.IsinRequest);
string? isin = req?.Isin;
string? savedPath = null;
if (!string.IsNullOrEmpty(isin))
{
savedPath = await indexService.DownloadAndSaveLogoAsync(isin);
}
string responseTopic = $"services/response/assets_FetchLogo/{correlationId}";
await PublishAsync(responseTopic, new FinlyticCore.Dtos.FetchLogoResponse(isin, savedPath, savedPath != null));
}
private async Task<List<AssetEntity>> HandleAssetsGetAsync(string payload, IAssetsDbService dbService)
{
var validReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetValidAssetRequest);
if (validReq != null)
{
return await dbService.GetValidAssetsByIsinAsync(validReq.Isin);
}
return [];
}
private async Task<List<AssetEntity>> HandleAssetsSearchAsync(string payload, IAssetsDbService dbService)
{
var searchReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.SearchAssetsRequest);
if (searchReq != null)
{
return await dbService.FindAffectedActiveAssetsAsync(searchReq.SearchQuery);
}
return [];
}
private async Task<List<AssetEntity>> HandleAssetsGetDiscoveryAsync(string payload, IAssetsDbService dbService)
{
int limit = 15;
if (!string.IsNullOrWhiteSpace(payload))
{
try
{
var discReq = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.GetDiscoveryAssetsRequest);
if (discReq != null && discReq.Limit > 0) limit = discReq.Limit;
}
catch { }
}
return await dbService.GetDiscoveryAssetsAsync(limit);
}
}
FinlyticAssets/Util/StringCodeGenerator.cs
+5 -2
View File
@@ -1,8 +1,11 @@
namespace FinlyticAssets.Util;
namespace FinlyticAssets.Util;
public class StringCodeGenerator
{
/// <summary>
/// Generates a W3C traceparent string for telemetry tracking.
/// </summary>
public static string GenerateTraceparent()
{
var traceId = Guid.NewGuid().ToString("N");
@@ -11,4 +14,4 @@ public class StringCodeGenerator
return $"00-{traceId}-{spanId}-01";
}
}
}
FinlyticAssets/Util/TradeRepublicClient.cs
-252
View File
@@ -1,252 +0,0 @@
using System.Collections.Concurrent;
using System.Text.Json;
using FinlyticAssets.Models.DataToObject.TradeRepublic;
namespace FinlyticAssets.Util;
/// <summary>
/// A managed WebSocket client designed to communicate with the Trade Republic API.
/// Handles asynchronous requests, generic serialization, and automatic subscription management.
/// </summary>
public class TradeRepublicClient : ManagedWebSocket
{
private readonly ILogger<TradeRepublicClient> _logger;
private int _currentSub;
private readonly ConcurrentDictionary<int, TaskCompletionSource<ReceivedMessage>> _pendingRequests = new();
/// <summary>
/// Wird ausgelöst, wenn Trade Republic asynchrone Updates (z.B. Live-Preise) schickt,
/// auf die niemand aktiv per SendRequestAsync wartet.
/// </summary>
public event Action<ReceivedMessage>? UnhandledMessageReceived;
/// <summary>
/// Wird ausgelöst, wenn Trade Republic Systemnachrichten oder Fehler ohne ID schickt.
/// </summary>
public event Action<string>? SystemMessageReceived;
/// <summary>
/// Initializes a new instance of the TradeRepublicClient.
/// Call InitAsync() afterwards to establish the connection.
/// </summary>
/// <param name="logger">The logger instance for tracking socket events and errors.</param>
public TradeRepublicClient(ILogger<TradeRepublicClient> logger)
{
_logger = logger;
}
/// <summary>
/// Asynchronously establishes the WebSocket connection to the Trade Republic API.
/// </summary>
public async Task<bool> InitAsync()
{
await ConnectAsync("wss://api.traderepublic.com/", TimeSpan.FromSeconds(10));
var tcs = new TaskCompletionSource<ReceivedMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
_pendingRequests.TryAdd(-1, tcs);
try
{
var json = JsonSerializer.Serialize(new TradeRepublicConnectRequest());
await SendAsync($"connect 34 {json}");
var res = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
if (string.IsNullOrWhiteSpace(res.Type)) return false;
var isConnected = res.Type == "connected";
if (isConnected)
{
_logger.LogInformation("WebSocket connection to Trade Republic established.");
}
return isConnected;
}
catch (TimeoutException)
{
_pendingRequests.TryRemove(-1, out _);
_logger.LogWarning("Timeout while waiting for response to ID {Id}.", -1);
return false;
}
catch (TaskCanceledException)
{
_logger.LogWarning(
"Trade Republic immediately rejected the request for ID {Id} (e.g., invalid ISIN or access denied).",
-1);
return false;
}
}
/// <summary>
/// Sends a strongly-typed request to the API and waits for the corresponding response.
/// Automatically handles the subscription ID and unsubscribes after completion or failure.
/// </summary>
/// <typeparam name="TResponse">The expected type of the response payload.</typeparam>
/// <typeparam name="TRequest">The type of the request payload.</typeparam>
/// <param name="request">The request data to be serialized and sent.</param>
/// <returns>The deserialized response object, or null if the request timed out or was canceled.</returns>
public async Task<TResponse?> SendRequestAsync<TResponse, TRequest>(TRequest request)
where TResponse : class where TRequest : class
{
var tempSub = Interlocked.Increment(ref _currentSub);
var msg = $"sub {tempSub} {JsonSerializer.Serialize(request)}";
Console.WriteLine(msg);
var tcs = new TaskCompletionSource<ReceivedMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
_pendingRequests.TryAdd(tempSub, tcs);
await SendAsync(msg);
try
{
var res = await tcs.Task.WaitAsync(TimeSpan.FromSeconds(5));
if (res.Type == null)
{
_logger.LogWarning("Trade Republic rejected the request for ID {Id} with message type '{Type}'.", tempSub, res.Type);
return null;
}
if (!res.Type.Contains('A'))
{
return null;
}
if (string.IsNullOrWhiteSpace(res.Data)) return null;
if (typeof(TResponse) == typeof(string))
{
return res.Data as TResponse;
}
else
{
return JsonSerializer.Deserialize<TResponse>(res.Data);
}
}
catch (TimeoutException)
{
_pendingRequests.TryRemove(tempSub, out _);
_logger.LogWarning("Timeout while waiting for response to ID {Id}.", tempSub);
return null;
}
catch (TaskCanceledException)
{
_logger.LogWarning(
"Trade Republic immediately rejected the request for ID {Id} (e.g., invalid ISIN or access denied).",
tempSub);
return null;
}
finally
{
if (IsConnected)
{
try
{
await SendAsync($"unsub {tempSub}");
}
catch
{
//ignore
}
}
}
}
/// <summary>
/// Processes incoming WebSocket messages, extracting the JSON payload and resolving pending tasks.
/// </summary>
/// <param name="message">The raw text message received from the server.</param>
protected override void OnMessageReceived(string message)
{
if (string.IsNullOrWhiteSpace(message)) return;
// 1. Handshake-Nachricht direkt abfangen
if (message == "connected")
{
if (_pendingRequests.TryRemove(-1, out var tcs))
{
tcs.SetResult(new ReceivedMessage(-1, "connected", null));
}
return;
}
// 2. Erstes Leerzeichen finden, um die ID zu isolieren
var firstSpaceIndex = message.IndexOf(' ');
if (firstSpaceIndex <= 0)
{
_logger.LogWarning("Unknown message format received: {Message}", message);
SystemMessageReceived?.Invoke(message);
return;
}
var idString = message[..firstSpaceIndex];
if (!int.TryParse(idString, out var responseId))
{
SystemMessageReceived?.Invoke(message);
return;
}
// Der Rest nach der ID (z. B. "A {...}" oder "C")
var remainder = message[firstSpaceIndex..].Trim();
// 3. Nachrichtentyp ("A", "C", etc.) und JSON-Inhalt sauber trennen
var nextSpaceIndex = remainder.IndexOf(' ');
string msgType;
string? json = null;
if (nextSpaceIndex == -1)
{
// Kein weiteres Leerzeichen vorhanden (wie bei "2 C")
msgType = remainder;
}
else
{
// Typ und JSON trennen (wie bei "2 A {...}")
msgType = remainder[..nextSpaceIndex].Trim();
json = remainder[nextSpaceIndex..].Trim();
}
// 4. KORREKTUR: "C" signalisiert nur das Ende des Datenstroms auf dieser ID.
// Wir ignorieren es, da die Daten bereits im Typ "A" übertragen wurden.
if (msgType == "C")
{
_logger.LogDebug("Trade Republic closed subscription channel for ID {ResponseId}.", responseId);
return;
}
// 5. Task auflösen, falls jemand auf diese ID wartet
if (_pendingRequests.TryRemove(responseId, out var pendingTcs))
{
pendingTcs.SetResult(new ReceivedMessage(responseId, msgType, json));
}
else
{
UnhandledMessageReceived?.Invoke(new ReceivedMessage(responseId, msgType, json));
}
}
/// <summary>
/// Determines whether the incoming message is a keep-alive echo response.
/// </summary>
/// <param name="message">The raw text message.</param>
/// <returns>True if the message is an echo response; otherwise, false.</returns>
protected override bool IsKeepAliveMessage(string message)
{
return message.StartsWith("echo");
}
/// <summary>
/// Sends a periodic keep-alive echo to maintain the WebSocket connection.
/// </summary>
protected override Task SendLifeMessageAsync()
{
var echo = $"echo {DateTimeOffset.UtcNow.ToUnixTimeSeconds()}";
return SendAsync(echo);
}
}
public record ReceivedMessage(int? Sub, string? Type, string? Data);
FinlyticAssets/Util/Volumes.cs
+7 -2
View File
@@ -1,4 +1,4 @@
namespace FinlyticAssets.Util;
namespace FinlyticAssets.Util;
public class Volumes
{
@@ -6,4 +6,9 @@ public class Volumes
/// Der relative Pfad für die schlanke Index-Datei (ISINs + Namen) zur Asset-Erkennung.
/// </summary>
public const string IndexRelativePath = "assets/index";
}
/// <summary>
/// Der relative Pfad für lokal gespeicherte Asset-Logos (nach ISIN benannt).
/// </summary>
public const string LogosRelativePath = "assets/logos";
}