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

feat(trades): dynamic settings, IFinlyticLogger, live log streaming, and EF migration

Browse Source
This commit is contained in:
Kleidukos
2026-08-15 21:30:19 +02:00
parent 0d370d09e7
commit 57554a9582
9 changed files with 588 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticTrades/Util/TradesMqttClient.cs
+105 -13
View File
@@ -6,9 +6,11 @@ using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using FinlyticCore.Dtos;
using FinlyticCore.Dtos.Settings;
using FinlyticCore.Dtos.TechnicalAnalysis;
using FinlyticCore.Models;
using FinlyticCore.Models.Trades;
using FinlyticCore.Services;
using FinlyticCore.Util;
using FinlyticTrades.Entities;
using FinlyticTrades.Services;
@@ -46,19 +48,19 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "finlytic_trades")}_{Guid.NewGuid():N}"
};
_logger.LogInformation("[{Channel}] Starting Unified Trades MQTT Client. Host: {Host}, ClientId: {ClientId}", "TradesChannel", config.Host, config.ClientId);
_logger.LogInformation("Starting Unified Trades MQTT Client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
await ConnectAsync(config);
}
public async Task StopAsync(CancellationToken cancellationToken)
{
_logger.LogInformation("[{Channel}] Stopping Unified Trades MQTT Client.", "TradesChannel");
_logger.LogInformation("Stopping Unified Trades MQTT Client.");
await DisconnectAsync();
}
protected override async Task OnConnectedAsync()
{
_logger.LogInformation("[{Channel}] Trades MQTT Client connected. Subscribing to topics...", "TradesChannel");
_logger.LogInformation("Trades MQTT Client connected. Subscribing to topics...");
await SubscribeAsync("finlytic/trades/proposed/#");
await SubscribeAsync("finlytic/trades/updates/#");
@@ -67,11 +69,21 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
await SubscribeAsync("services/request/trades_Close/#");
await SubscribeAsync("services/request/trades_Reject/#");
await SubscribeAsync("services/request/trades_Accept/#");
await SubscribeAsync("services/request/trades_settings_GetAll/#");
await SubscribeAsync("services/request/trades_settings_Update/#");
await SubscribeAsync("services/config/updated/#");
await SubscribeAsync("services/request/health_Ping/#");
await SubscribeAsync("services/response/tr_GetLivePrice/#");
_logger.LogInformation("[{Channel}] Successfully subscribed to all event and RPC channels.", "TradesChannel");
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
{
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticTrades", StringComparison.OrdinalIgnoreCase))
{
await PublishAsync("finlytic/logs/FinlyticTrades", logDto);
}
};
_logger.LogInformation("Successfully subscribed to all event and RPC channels.");
}
protected override async Task OnMessageReceivedAsync(string topic, string payloadStr)
@@ -91,7 +103,9 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
string respTopic = $"services/response/health_Ping/{correlationId}";
var healthResp = new ServiceHealthResponse("FinlyticTrades", "Online", DateTime.UtcNow, "Connected");
await PublishAsync(respTopic, healthResp);
_logger.LogInformation("[{Channel}] [TradesMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", "TradesChannel", correlationId);
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TradesMqttClient>>();
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[TradesMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
}
return;
}
@@ -100,22 +114,35 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
{
if (topic.EndsWith("FinlyticTrades", StringComparison.OrdinalIgnoreCase))
{
_logger.LogInformation("[{Channel}] [TradesMqttClient] Received config update event for FinlyticTrades.", "TradesChannel");
var payload = JsonSerializer.Deserialize(payloadStr, FinlyticJsonSerializerContext.Default.ServiceConfigUpdatePayload);
if (payload?.Settings != null && payload.Settings.Count > 0)
{
using var scope = _scopeFactory.CreateScope();
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
await settingsDb.UpdateSettingsFromDictionaryAsync(payload.Settings);
_logger.LogInformation("[{Channel}] [TradesMqttClient] Persisted {Count} updated settings to FinlyticTrades database.", "TradesChannel", payload.Settings.Count);
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
var dict = payload.Settings.ToDictionary(k => k.Key, v => (object?)v.Value);
await settings.UpdateSettingsAsync(dict);
}
}
return;
}
// Für Scoped-Services erzeugen wir pro eingehender Nachricht einen eigenen Scope
if (topic.StartsWith("services/request/trades_settings_GetAll", StringComparison.OrdinalIgnoreCase))
{
var correlationId = topic.Split('/').Last();
await HandleSettingsGetAllAsync(correlationId);
return;
}
if (topic.StartsWith("services/request/trades_settings_Update", StringComparison.OrdinalIgnoreCase))
{
var correlationId = topic.Split('/').Last();
await HandleSettingsUpdateAsync(payloadStr, correlationId);
return;
}
using var msgScope = _scopeFactory.CreateScope();
var tradeLifecycleService = msgScope.ServiceProvider.GetRequiredService<ITradeLifecycleService>();
var finlyticLoggerInstance = msgScope.ServiceProvider.GetRequiredService<IFinlyticLogger<TradesMqttClient>>();
if (topic.StartsWith("finlytic/trades/proposed/"))
{
@@ -126,7 +153,7 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
}
else
{
_logger.LogWarning("[{Channel}] [TradesMqttClient] Received proposed trade payload but Symbol/ISIN is empty. Skipping ingestion.", "TradesChannel");
await finlyticLoggerInstance.LogWarningAsync(SettingKeys.TradesChannel, "[TradesMqttClient] Received proposed trade payload but Symbol/ISIN is empty. Skipping ingestion.");
}
}
else if (topic.StartsWith("finlytic/trades/accept/"))
@@ -199,7 +226,7 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
}
catch (Exception ex)
{
_logger.LogDebug(ex, "[{Channel}] Live price fetch skipped or timed out during trades_Get", "TradesChannel");
await finlyticLoggerInstance.LogDebugAsync(SettingKeys.TradesChannel, "[TradesMqttClient] Live price fetch skipped or timed out during trades_Get: {Message}", ex.Message);
}
}
@@ -250,7 +277,72 @@ public class TradesMqttClient : ManagedMqttClient, IHostedService
}
catch (Exception ex)
{
_logger.LogError(ex, "[{Channel}] Error processing incoming MQTT message on topic {Topic}", "TradesChannel", topic);
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TradesMqttClient>>();
await finlyticLogger.LogErrorAsync(SettingKeys.TradesChannel, ex, "[TradesMqttClient] Error processing incoming MQTT message on topic {Topic}", topic);
}
}
private async Task HandleSettingsGetAllAsync(string correlationId)
{
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TradesMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTrades] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
try
{
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
var responseTopic = $"services/response/trades_settings_GetAll/{correlationId}";
await PublishAsync(responseTopic, settings);
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTrades] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
}
catch (Exception ex)
{
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticTrades] [Settings_GetAll] Failed to retrieve settings.");
}
}
private async Task HandleSettingsUpdateAsync(string payload, string correlationId)
{
if (string.IsNullOrWhiteSpace(payload)) return;
using var scope = _scopeFactory.CreateScope();
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<TradesMqttClient>>();
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTrades] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
try
{
Dictionary<string, object?>? updates = null;
try
{
updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
}
catch
{
var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
if (list != null)
{
updates = new Dictionary<string, object?>();
foreach (var item in list) updates[item.Key] = item.Value;
}
}
if (updates != null && updates.Count > 0)
{
await settingsService.UpdateSettingsAsync(updates);
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticTrades] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
}
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
var responseTopic = $"services/response/trades_settings_Update/{correlationId}";
await PublishAsync(responseTopic, currentSettings);
}
catch (Exception ex)
{
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticTrades] [Settings_Update] Failed to update settings.");
}
}