From 1f9d66405a19c9fdd449fdd3e7832998a2dce48c Mon Sep 17 00:00:00 2001 From: Kleidukos Date: Sat, 15 Aug 2026 21:30:46 +0200 Subject: [PATCH] feat(assets): dynamic settings, IFinlyticLogger, live log streaming, and EF migration --- FinlyticAssets/Database/AssetsDbContext.cs | 20 +- ...dateDynamicSettingsUniqueIndex.Designer.cs | 366 +++++++++++++ ...184053_UpdateDynamicSettingsUniqueIndex.cs | 37 ++ .../AssetsDbContextModelSnapshot.cs | 3 +- FinlyticAssets/Program.cs | 25 +- .../Services/AssetScannerBackgroundService.cs | 56 +- FinlyticAssets/Services/AssetsDbService.cs | 479 ++++++++---------- FinlyticAssets/Services/AssetsIndexService.cs | 38 +- .../Services/LogoFetcherBackgroundService.cs | 32 +- FinlyticAssets/Util/AssetsMqttClient.cs | 161 ++++-- FinlyticAssets/Util/SettingKeys.cs | 22 + 11 files changed, 855 insertions(+), 384 deletions(-) create mode 100644 FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.Designer.cs create mode 100644 FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.cs create mode 100644 FinlyticAssets/Util/SettingKeys.cs diff --git a/FinlyticAssets/Database/AssetsDbContext.cs b/FinlyticAssets/Database/AssetsDbContext.cs index 3213c31..0603190 100644 --- a/FinlyticAssets/Database/AssetsDbContext.cs +++ b/FinlyticAssets/Database/AssetsDbContext.cs @@ -1,11 +1,16 @@ +using System; +using System.Collections.Generic; +using System.Linq; using FinlyticAssets.Entities; +using FinlyticCore.Database; using FinlyticCore.Entities.Settings; using Microsoft.EntityFrameworkCore; using Microsoft.EntityFrameworkCore.ChangeTracking; +using Microsoft.EntityFrameworkCore.Design; namespace FinlyticAssets.Database; -public class AssetsDbContext : DbContext +public class AssetsDbContext : DbContext, ISettingsDbContext { public AssetsDbContext(DbContextOptions options) : base(options) { @@ -16,7 +21,6 @@ public class AssetsDbContext : DbContext public DbSet TradeRepublicAssets { get; set; } public DbSet TradeRepublicTags { get; set; } - protected override void OnModelCreating(ModelBuilder modelBuilder) { base.OnModelCreating(modelBuilder); @@ -24,7 +28,7 @@ public class AssetsDbContext : DbContext modelBuilder.Entity(entity => { entity.HasKey(e => e.Id); - entity.HasIndex(e => e.Key); + entity.HasIndex(e => e.Key).IsUnique(); }); modelBuilder.Entity(entity => { @@ -95,3 +99,13 @@ public class AssetsDbContext : DbContext .HaveConversion(typeof(FinlyticCore.Converters.NullableUtcDateTimeConverter)); } } + +public class AssetsDbContextFactory : IDesignTimeDbContextFactory +{ + public AssetsDbContext CreateDbContext(string[] args) + { + var optionsBuilder = new DbContextOptionsBuilder(); + optionsBuilder.UseNpgsql("Host=localhost;Database=assets;Username=postgres;Password=postgres"); + return new AssetsDbContext(optionsBuilder.Options); + } +} diff --git a/FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.Designer.cs b/FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.Designer.cs new file mode 100644 index 0000000..e7c5aa2 --- /dev/null +++ b/FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.Designer.cs @@ -0,0 +1,366 @@ +// +using System; +using FinlyticAssets.Database; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; + +#nullable disable + +namespace FinlyticAssets.Migrations +{ + [DbContext(typeof(AssetsDbContext))] + [Migration("20260815184053_UpdateDynamicSettingsUniqueIndex")] + partial class UpdateDynamicSettingsUniqueIndex + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.9") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("AssetEntityTagEntity", b => + { + b.Property("TagsId") + .HasColumnType("text"); + + b.Property("AssetsIsin") + .HasColumnType("text"); + + b.Property("AssetsInstrumentCategory") + .HasColumnType("text"); + + b.HasKey("TagsId", "AssetsIsin", "AssetsInstrumentCategory"); + + b.HasIndex("AssetsIsin", "AssetsInstrumentCategory"); + + b.ToTable("AssetEntityTagEntity"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.AssetEntity", b => + { + b.Property("Isin") + .HasColumnType("text"); + + b.Property("InstrumentCategory") + .HasColumnType("text"); + + b.Property("AssetType") + .IsRequired() + .HasMaxLength(13) + .HasColumnType("character varying(13)"); + + b.Property("HasCfd") + .HasColumnType("boolean"); + + b.Property("ImageId") + .HasColumnType("text"); + + b.Property("LastUpdatedAt") + .HasColumnType("timestamp with time zone"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text"); + + b.Property("Type") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Isin", "InstrumentCategory"); + + b.HasIndex("LastUpdatedAt"); + + b.ToTable("TradeRepublicAssets"); + + b.HasDiscriminator("AssetType").HasValue("AssetEntity"); + + b.UseTphMappingStrategy(); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.Settings", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("AssetUpdateTypeDelay") + .HasColumnType("integer"); + + b.Property("BatchAssetUpdateDelay") + .HasColumnType("integer"); + + b.Property("CurrentScanningPage") + .HasColumnType("integer"); + + b.Property("CurrentScanningType") + .IsRequired() + .HasMaxLength(50) + .HasColumnType("character varying(50)"); + + b.Property("FinishedInitialScan") + .HasColumnType("boolean"); + + b.Property("InitAssetUpdateTypeDelay") + .HasColumnType("integer"); + + b.Property("InitBatchAssetUpdateDelay") + .HasColumnType("integer"); + + b.Property("TradeRepublicMaxRequestPageSize") + .HasColumnType("integer"); + + b.HasKey("Id"); + + b.ToTable("Settings"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.TagEntity", b => + { + b.Property("Id") + .HasColumnType("text"); + + b.Property("Name") + .IsRequired() + .HasColumnType("text"); + + b.Property("Type") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.ToTable("TradeRepublicTags"); + }); + + modelBuilder.Entity("FinlyticCore.Entities.Settings.SettingEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("uuid"); + + b.Property("Key") + .IsRequired() + .HasMaxLength(150) + .HasColumnType("character varying(150)"); + + b.Property("LastUpdatedUtc") + .HasColumnType("timestamp with time zone"); + + b.Property("ServiceIdentifier") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("ValueJson") + .IsRequired() + .HasColumnType("text"); + + b.HasKey("Id"); + + b.HasIndex("Key") + .IsUnique(); + + b.ToTable("DynamicSettings"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.BondEntity", b => + { + b.HasBaseType("FinlyticAssets.Entities.AssetEntity"); + + b.Property("BondIssuerName") + .IsRequired() + .HasColumnType("text"); + + b.Property("SearchSubtitle") + .IsRequired() + .HasColumnType("text"); + + b.HasDiscriminator().HasValue("Bond"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.CryptoEntity", b => + { + b.HasBaseType("FinlyticAssets.Entities.AssetEntity"); + + b.Property("SearchSubtitle") + .IsRequired() + .HasColumnType("text"); + + b.Property("Subtitle") + .IsRequired() + .HasColumnType("text"); + + b.ToTable("TradeRepublicAssets", t => + { + t.Property("SearchSubtitle") + .HasColumnName("CryptoEntity_SearchSubtitle"); + }); + + b.HasDiscriminator().HasValue("Crypto"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.DerivativeEntity", b => + { + b.HasBaseType("FinlyticAssets.Entities.AssetEntity"); + + b.Property("Barrier") + .HasColumnType("numeric(18,6)"); + + b.Property("Currency") + .IsRequired() + .HasMaxLength(10) + .HasColumnType("character varying(10)"); + + b.Property("Delta") + .HasColumnType("numeric"); + + b.Property("DerivativeProductCategories") + .IsRequired() + .HasColumnType("text"); + + b.Property("Expiry") + .HasColumnType("timestamp with time zone"); + + b.Property("Factor") + .HasColumnType("numeric"); + + b.Property("Issuer") + .IsRequired() + .HasMaxLength(150) + .HasColumnType("character varying(150)"); + + b.Property("IssuerDisplayName") + .IsRequired() + .HasMaxLength(150) + .HasColumnType("character varying(150)"); + + b.Property("IssuerImageId") + .HasColumnType("text"); + + b.Property("Leverage") + .HasColumnType("numeric(10,4)"); + + b.Property("NextGenProductCategoryName") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("OptionType") + .HasColumnType("integer"); + + b.Property("ProductCategoryName") + .IsRequired() + .HasMaxLength(100) + .HasColumnType("character varying(100)"); + + b.Property("Size") + .HasColumnType("numeric"); + + b.Property("Strike") + .HasColumnType("numeric(18,6)"); + + b.Property("UnderlyingIsin") + .HasMaxLength(12) + .HasColumnType("character varying(12)"); + + b.HasDiscriminator().HasValue("Derivative"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.EtfEntity", b => + { + b.HasBaseType("FinlyticAssets.Entities.AssetEntity"); + + b.Property("DerivativeProductCategories") + .IsRequired() + .HasColumnType("text"); + + b.Property("EtfDescription") + .IsRequired() + .HasColumnType("text"); + + b.Property("MappedEtfIndexName") + .IsRequired() + .HasColumnType("text"); + + b.Property("SearchSubtitle") + .IsRequired() + .HasColumnType("text"); + + b.Property("Subtitle") + .IsRequired() + .HasColumnType("text"); + + b.ToTable("TradeRepublicAssets", t => + { + t.Property("DerivativeProductCategories") + .HasColumnName("EtfEntity_DerivativeProductCategories"); + + t.Property("SearchSubtitle") + .HasColumnName("EtfEntity_SearchSubtitle"); + + t.Property("Subtitle") + .HasColumnName("EtfEntity_Subtitle"); + }); + + b.HasDiscriminator().HasValue("Etf"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.StockEntity", b => + { + b.HasBaseType("FinlyticAssets.Entities.AssetEntity"); + + b.Property("DerivativeProductCategories") + .IsRequired() + .HasColumnType("text"); + + b.ToTable("TradeRepublicAssets", t => + { + t.Property("DerivativeProductCategories") + .HasColumnName("StockEntity_DerivativeProductCategories"); + }); + + b.HasDiscriminator().HasValue("Stock"); + }); + + modelBuilder.Entity("FinlyticAssets.Entities.SyntheticEntity", b => + { + b.HasBaseType("FinlyticAssets.Entities.AssetEntity"); + + b.Property("DerivativeProductCategories") + .IsRequired() + .HasColumnType("text"); + + b.ToTable("TradeRepublicAssets", t => + { + t.Property("DerivativeProductCategories") + .HasColumnName("SyntheticEntity_DerivativeProductCategories"); + }); + + b.HasDiscriminator().HasValue("Synthetic"); + }); + + modelBuilder.Entity("AssetEntityTagEntity", b => + { + b.HasOne("FinlyticAssets.Entities.TagEntity", null) + .WithMany() + .HasForeignKey("TagsId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + + b.HasOne("FinlyticAssets.Entities.AssetEntity", null) + .WithMany() + .HasForeignKey("AssetsIsin", "AssetsInstrumentCategory") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.cs b/FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.cs new file mode 100644 index 0000000..025e9a2 --- /dev/null +++ b/FinlyticAssets/Migrations/20260815184053_UpdateDynamicSettingsUniqueIndex.cs @@ -0,0 +1,37 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace FinlyticAssets.Migrations +{ + /// + public partial class UpdateDynamicSettingsUniqueIndex : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_DynamicSettings_Key", + table: "DynamicSettings"); + + migrationBuilder.CreateIndex( + name: "IX_DynamicSettings_Key", + table: "DynamicSettings", + column: "Key", + unique: true); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_DynamicSettings_Key", + table: "DynamicSettings"); + + migrationBuilder.CreateIndex( + name: "IX_DynamicSettings_Key", + table: "DynamicSettings", + column: "Key"); + } + } +} diff --git a/FinlyticAssets/Migrations/AssetsDbContextModelSnapshot.cs b/FinlyticAssets/Migrations/AssetsDbContextModelSnapshot.cs index 9294699..26ad3b5 100644 --- a/FinlyticAssets/Migrations/AssetsDbContextModelSnapshot.cs +++ b/FinlyticAssets/Migrations/AssetsDbContextModelSnapshot.cs @@ -161,7 +161,8 @@ namespace FinlyticAssets.Migrations b.HasKey("Id"); - b.HasIndex("Key"); + b.HasIndex("Key") + .IsUnique(); b.ToTable("DynamicSettings"); }); diff --git a/FinlyticAssets/Program.cs b/FinlyticAssets/Program.cs index 599d7ef..2b98df0 100644 --- a/FinlyticAssets/Program.cs +++ b/FinlyticAssets/Program.cs @@ -1,19 +1,27 @@ -using System.Text.Json; -using FinlyticAssets; +using System; using FinlyticAssets.Database; -using FinlyticCore.Services.TradeRepublic; -using FinlyticAssets.Util; using FinlyticAssets.Services; +using FinlyticAssets.Util; +using FinlyticCore.Database; +using FinlyticCore.Services; +using FinlyticCore.Services.TradeRepublic; using Microsoft.EntityFrameworkCore; -using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; var builder = Host.CreateApplicationBuilder(args); +// DB Context builder.Services.AddDbContext(options => options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))); +builder.Services.AddScoped(sp => sp.GetRequiredService()); + +// Core Services +builder.Services.AddSingleton(); +builder.Services.AddSingleton(typeof(IFinlyticLogger<>), typeof(FinlyticLogger<>)); builder.Services.AddScoped(); - builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); @@ -34,8 +42,6 @@ using (var scope = host.Services.CreateScope()) var context = scope.ServiceProvider.GetRequiredService(); await context.Database.MigrateAsync(); - // Fix: Reset InitAssetUpdateTypeDelay from old default (3600s) to new default (0s = no delay). - // This ensures the scanner moves immediately to the next asset type during the initial scan. var settingsService = scope.ServiceProvider.GetRequiredService(); var settings = await settingsService.GetSettings(); if (settings.InitAssetUpdateTypeDelay == 3600) @@ -47,8 +53,7 @@ using (var scope = host.Services.CreateScope()) } catch (Exception ex) { - Console.WriteLine($"Critical error during database migration: {ex.Message}"); - Console.WriteLine(ex.StackTrace); + Console.WriteLine($"Critical error during database migration for FinlyticAssets: {ex.Message}"); } } diff --git a/FinlyticAssets/Services/AssetScannerBackgroundService.cs b/FinlyticAssets/Services/AssetScannerBackgroundService.cs index 7e4d035..ed57968 100644 --- a/FinlyticAssets/Services/AssetScannerBackgroundService.cs +++ b/FinlyticAssets/Services/AssetScannerBackgroundService.cs @@ -4,12 +4,13 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; using FinlyticAssets.Models; +using FinlyticAssets.Util; using FinlyticCore.Dtos.TradeRepublic; using FinlyticCore.Models.Assets; +using FinlyticCore.Services; using FinlyticCore.Services.TradeRepublic; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; namespace FinlyticAssets.Services; @@ -20,31 +21,31 @@ namespace FinlyticAssets.Services; public class AssetScannerBackgroundService : BackgroundService { private readonly IServiceScopeFactory _serviceScopeFactory; - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; private AssetsCount? _assetsCount; private AssetsCount? _currAssetsCount; - public AssetScannerBackgroundService(IServiceScopeFactory serviceScopeFactory, ILogger logger) + public AssetScannerBackgroundService(IServiceScopeFactory serviceScopeFactory, IFinlyticLogger finlyticLogger) { _serviceScopeFactory = serviceScopeFactory; - _logger = logger; + _finlyticLogger = finlyticLogger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - _logger.LogInformation("[{Channel}] AssetScannerBackgroundService has started.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] AssetScannerBackgroundService has started."); try { using var scope = _serviceScopeFactory.CreateScope(); var indexService = scope.ServiceProvider.GetRequiredService(); - _logger.LogInformation("[{Channel}] Building initial asset index on service startup...", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Building initial asset index on service startup..."); await indexService.ReCreateIndexFileAsync(stoppingToken); } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] Failed to build initial asset index on startup. Continuing service execution.", "AssetsChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetScannerBackgroundService] Failed to build initial asset index on startup. Continuing service execution."); } do @@ -57,7 +58,7 @@ public class AssetScannerBackgroundService : BackgroundService var assetsDbService = scope.ServiceProvider.GetRequiredService(); var indexService = scope.ServiceProvider.GetRequiredService(); - _logger.LogInformation("[{Channel}] Requesting total asset counts from Trade Republic...", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Requesting total asset counts from Trade Republic..."); _assetsCount = await tradeRepublicService.GetAssetsCount(stoppingToken); _currAssetsCount = new AssetsCount(); @@ -72,7 +73,7 @@ public class AssetScannerBackgroundService : BackgroundService { if (type != initSettings.CurrentScanningType) { - _logger.LogInformation("[{Channel}] Recovery: {AssetType} was already processed. Skipping.", "AssetsChannel", type); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Recovery: {AssetType} was already processed. Skipping.", type); continue; } isRecoveryMode = false; @@ -85,7 +86,7 @@ public class AssetScannerBackgroundService : BackgroundService await settingsService.SaveSettings(settings); } - _logger.LogInformation("[{Channel}] Processing asset type: {AssetType}...", "AssetsChannel", type); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Processing asset type: {AssetType}...", type); await HandleAssetType(type, tradeRepublicService, settingsService, assetsDbService, indexService, stoppingToken); var currentSettings = await settingsService.GetSettings(); @@ -96,7 +97,7 @@ public class AssetScannerBackgroundService : BackgroundService if (delaySeconds > 0) { var jitter = Random.Shared.Next(0, Math.Min(15, delaySeconds)); - _logger.LogInformation("[{Channel}] Waiting {Delay}s before next asset type ({Type}).", "AssetsChannel", delaySeconds + jitter, type); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Waiting {Delay}s before next asset type ({Type}).", delaySeconds + jitter, type); await Task.Delay(TimeSpan.FromSeconds(delaySeconds + jitter), stoppingToken); } } @@ -106,23 +107,23 @@ public class AssetScannerBackgroundService : BackgroundService if (!finalSettings.FinishedInitialScan && !stoppingToken.IsCancellationRequested) { - _logger.LogInformation("[{Channel}] Initial scan successfully completed. Switching FinishedInitialScan to true.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Initial scan successfully completed. Switching FinishedInitialScan to true."); finalSettings.FinishedInitialScan = true; } await settingsService.SaveSettings(finalSettings); - _logger.LogInformation("[{Channel}] Full scan cycle completed. Waiting 1 minute before starting the next cycle.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Full scan cycle completed. Waiting 1 minute before starting the next cycle."); await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken); } catch (Exception e) when (!stoppingToken.IsCancellationRequested) { - _logger.LogError(e, "[{Channel}] An unhandled exception occurred in AssetScannerBackgroundService. Retrying in 10 seconds.", "AssetsChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, e, "[AssetScannerBackgroundService] An unhandled exception occurred in AssetScannerBackgroundService. Retrying in 10 seconds."); try { await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken); } catch { /* Ignore */ } } } while (!stoppingToken.IsCancellationRequested); - _logger.LogInformation("[{Channel}] AssetScannerBackgroundService is stopping.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] AssetScannerBackgroundService is stopping."); } private async Task HandleAssetType( @@ -136,7 +137,7 @@ public class AssetScannerBackgroundService : BackgroundService var totalCount = _assetsCount?.GetCountFromType(type) ?? 0; if (totalCount == 0) { - _logger.LogWarning("[{Channel}] No assets found for type {AssetType}.", "AssetsChannel", type); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] No assets found for type {AssetType}.", type); return; } @@ -149,8 +150,8 @@ public class AssetScannerBackgroundService : BackgroundService if (settings.CurrentScanningType == type && settings.CurrentScanningPage > 0) { currentItemOffset = (settings.CurrentScanningPage - 1) * pageSize; - _logger.LogInformation("[{Channel}] Resuming full scan for {AssetType} from Page {Page} (Calculated Offset: {Offset}).", - "AssetsChannel", type, settings.CurrentScanningPage, currentItemOffset); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Resuming full scan for {AssetType} from Page {Page} (Calculated Offset: {Offset}).", + type, settings.CurrentScanningPage, currentItemOffset); } while (currentItemOffset < totalCount && !stoppingToken.IsCancellationRequested) @@ -164,15 +165,14 @@ public class AssetScannerBackgroundService : BackgroundService currentSettings.CurrentScanningPage = currentPage; await settingsDbService.SaveSettings(currentSettings); - _logger.LogDebug("Fetching {AssetType} - Page {Page}. Numerical Offset: {Offset}/{Total}", + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Fetching {AssetType} - Page {Page}. Numerical Offset: {Offset}/{Total}", type, currentPage, currentItemOffset, totalCount); var assets = await tradeRepublicService.GetAssets(type, currentPage, pageSize, stoppingToken); - // Keine Ergebnisse geliefert -> Katalogende erreicht if (assets?.Results == null || assets.Results.Count == 0) { - _logger.LogInformation("[{Channel}] Fetch for {AssetType} (Page {Page}) returned no results. Reached end of available assets.", "AssetsChannel", type, currentPage); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Fetch for {AssetType} (Page {Page}) returned no results. Reached end of available assets.", type, currentPage); currentSettings.CurrentScanningPage = 0; await settingsDbService.SaveSettings(currentSettings); break; @@ -183,10 +183,9 @@ public class AssetScannerBackgroundService : BackgroundService currentItemOffset += assets.Results.Count; - // Unvollständige Seite -> Letzte Seite abgearbeitet if (assets.Results.Count < pageSize) { - _logger.LogInformation("[{Channel}] Reached the last page for {AssetType}.", "AssetsChannel", type); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Reached the last page for {AssetType}.", type); currentSettings.CurrentScanningPage = 0; await settingsDbService.SaveSettings(currentSettings); break; @@ -198,16 +197,15 @@ public class AssetScannerBackgroundService : BackgroundService if (delaySeconds > 0) { - // Angemessener Jitter (0 bis max. 5 Sek. bzw. kleiner als delaySeconds) var maxJitter = Math.Min(5, delaySeconds); var jitter = Random.Shared.Next(0, maxJitter + 1); - _logger.LogDebug("Waiting {Delay} seconds before the next batch.", delaySeconds + jitter); + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Waiting {Delay} seconds before the next batch.", delaySeconds + jitter); await Task.Delay(TimeSpan.FromSeconds(delaySeconds + jitter), stoppingToken); } } - _logger.LogInformation("[{Channel}] Finished scanning {AssetType}. Total scanned in this cycle: {Count}/{Total}", - "AssetsChannel", type, _currAssetsCount.GetCountFromType(type), totalCount); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Finished scanning {AssetType}. Total scanned in this cycle: {Count}/{Total}", + type, _currAssetsCount.GetCountFromType(type), totalCount); } private async Task ProcessAssets( @@ -219,11 +217,11 @@ public class AssetScannerBackgroundService : BackgroundService if (assets == null || assets.Count == 0) return; var changedRows = await assetsDbService.AddOrUpdateAssetsAsync(assets); - _logger.LogInformation("[{Channel}] [Scan] {Count} assets passed to the DB service. {Changed} modifications/inserts executed.", "AssetsChannel", assets.Count, changedRows); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] [Scan] {Count} assets passed to the DB service. {Changed} modifications/inserts executed.", assets.Count, changedRows); if (changedRows > 0) { - _logger.LogInformation("[{Channel}] Database modifications detected. Recreating the asset index file...", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetScannerBackgroundService] Database modifications detected. Recreating the asset index file..."); await indexService.ReCreateIndexFileAsync(stoppingToken); } } diff --git a/FinlyticAssets/Services/AssetsDbService.cs b/FinlyticAssets/Services/AssetsDbService.cs index 90e7db6..7c92892 100644 --- a/FinlyticAssets/Services/AssetsDbService.cs +++ b/FinlyticAssets/Services/AssetsDbService.cs @@ -1,6 +1,13 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; using FinlyticAssets.Database; using FinlyticAssets.Entities; +using FinlyticAssets.Util; using FinlyticCore.Dtos.TradeRepublic; +using FinlyticCore.Services; using FinlyticCore.Services.TradeRepublic; using Microsoft.EntityFrameworkCore; @@ -28,12 +35,12 @@ public class AssetsDbService : IAssetsDbService { private readonly AssetsDbContext _context; private readonly ITradeRepublicService _tradeRepublicService; - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; - public AssetsDbService(AssetsDbContext context, ILogger logger, ITradeRepublicService tradeRepublicService) + public AssetsDbService(AssetsDbContext context, IFinlyticLogger finlyticLogger, ITradeRepublicService tradeRepublicService) { _context = context; - _logger = logger; + _finlyticLogger = finlyticLogger; _tradeRepublicService = tradeRepublicService; } @@ -60,42 +67,41 @@ public class AssetsDbService : IAssetsDbService + (a.Name.Length > 3 ? 5 : 0) }) .OrderByDescending(x => x.Score) - .ThenByDescending(x => x.Asset.LastUpdatedAt) + .Take(limit) + .Select(x => x.Asset) .ToList(); - var result = new List(); - var grouped = scored.GroupBy(x => x.Asset.Type).ToList(); - - int index = 0; - while (result.Count < limit && grouped.Any(g => g.Any())) - { - bool addedAny = false; - foreach (var group in grouped) - { - var item = group.Skip(index).FirstOrDefault(); - if (item != null) - { - result.Add(item.Asset); - addedAny = true; - if (result.Count >= limit) break; - } - } - index++; - if (!addedAny) break; - } - - return result; + return scored; } /// Inherits documentation from interface. public async Task> GetAllValidAssetsAsync() { var cutoff = DateTime.UtcNow.AddDays(-90); - + return await _context.TradeRepublicAssets + .AsNoTracking() + .Where(a => a.LastUpdatedAt >= cutoff) + .ToListAsync(); + } + + /// Inherits documentation from interface. + public async Task> GetAssetsByIsinAsync(string isin) + { return await _context.TradeRepublicAssets .AsNoTracking() .Include(a => a.Tags) - .Where(a => a.LastUpdatedAt >= cutoff) + .Where(a => a.Isin == isin) + .ToListAsync(); + } + + /// Inherits documentation from interface. + public async Task> GetValidAssetsByIsinAsync(string isin) + { + var cutoff = DateTime.UtcNow.AddDays(-90); + return await _context.TradeRepublicAssets + .AsNoTracking() + .Include(a => a.Tags) + .Where(a => a.Isin == isin && a.LastUpdatedAt >= cutoff) .ToListAsync(); } @@ -114,7 +120,7 @@ public class AssetsDbService : IAssetsDbService var existingTag = await _context.TradeRepublicTags.FirstOrDefaultAsync(t => t.Id == tagDto.Id); if (existingTag == null) { - _logger.LogTrace("Creating missing asset tag object in storage cache context. ID: {TagId}", tagDto.Id); + await _finlyticLogger.LogTraceAsync(SettingKeys.AssetsChannel, "Creating missing asset tag object in storage cache context. ID: {TagId}", tagDto.Id); existingTag = new TagEntity { Id = tagDto.Id, Name = tagDto.Name, Type = tagDto.Type }; await _context.TradeRepublicTags.AddAsync(existingTag); } @@ -124,7 +130,7 @@ public class AssetsDbService : IAssetsDbService if (existingEntity == null) { - _logger.LogDebug("Asset with ISIN {Isin} not found. Mapping and creating a new entity.", dtoAsset.Isin); + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "Asset with ISIN {Isin} not found. Mapping and creating a new entity.", dtoAsset.Isin); var newEntity = MapDtoToEntity(dtoAsset); newEntity.LastUpdatedAt = now; @@ -135,7 +141,7 @@ public class AssetsDbService : IAssetsDbService return true; } - _logger.LogDebug("Asset with ISIN {Isin} exists. Merging properties and updating database record.", dtoAsset.Isin); + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "Asset with ISIN {Isin} exists. Merging properties and updating database record.", dtoAsset.Isin); existingEntity.Name = dtoAsset.Name; existingEntity.Type = dtoAsset.Type; @@ -187,7 +193,7 @@ public class AssetsDbService : IAssetsDbService { if (!tagCache.TryGetValue(tagDto.Id, out var tagEntity)) { - _logger.LogTrace("Creating missing asset tag object in storage cache context. ID: {TagId}", tagDto.Id); + await _finlyticLogger.LogTraceAsync(SettingKeys.AssetsChannel, "Creating missing asset tag object in storage cache context. ID: {TagId}", tagDto.Id); tagEntity = new TagEntity { Id = tagDto.Id, Name = tagDto.Name, Type = tagDto.Type }; await _context.TradeRepublicTags.AddAsync(tagEntity); tagCache.Add(tagDto.Id, tagEntity); @@ -197,7 +203,7 @@ public class AssetsDbService : IAssetsDbService if (existingEntity == null) { - _logger.LogDebug("Asset with ISIN {Isin} not found. Mapping and creating a new entity.", dto.Isin); + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "Asset with ISIN {Isin} not found. Mapping and creating a new entity.", dto.Isin); var newEntity = MapDtoToEntity(dto); newEntity.LastUpdatedAt = now; @@ -208,7 +214,7 @@ public class AssetsDbService : IAssetsDbService } else { - _logger.LogDebug("Asset with ISIN {Isin} exists. Merging properties and updating database record.", dto.Isin); + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "Asset with ISIN {Isin} exists. Merging properties and updating database record.", dto.Isin); if (existingEntity.Name != dto.Name || existingEntity.Type != dto.Type || @@ -223,140 +229,137 @@ public class AssetsDbService : IAssetsDbService existingEntity.HasCfd = dto.HasCfd; existingEntity.ImageId = dto.ImageId; existingEntity.LastUpdatedAt = now; - - UpdateSubtypeProperties(existingEntity, dto); existingEntity.Tags = mappedTags; + UpdateSubtypeProperties(existingEntity, dto); _context.TradeRepublicAssets.Update(existingEntity); isChanged = true; } } - if (isChanged) - { - changedCount++; - } - } - - if (changedCount > 0) - { - await _context.SaveChangesAsync(); + if (isChanged) changedCount++; } + await _context.SaveChangesAsync(); return changedCount; } - /// Inherits documentation from interface. - public async Task> GetAssetsByIsinAsync(string isin) + private static AssetEntity MapDtoToEntity(TradeRepublicAsset dto) { - var localAssets = await _context.TradeRepublicAssets - .Include(a => a.Tags) - .Where(a => a.Isin == isin) - .ToListAsync(); - - if (localAssets.Count > 0) + return dto switch { - return localAssets; - } - - // JIT-Fetch via API - var trAssetDto = await _tradeRepublicService.GetAsset(isin); - if (trAssetDto?.Results != null && trAssetDto.Results.Count > 0) - { - foreach (var asset in trAssetDto.Results) + TradeRepublicStock stock => new StockEntity { - await AddOrUpdateAssetAsync(asset); + Isin = stock.Isin, + Name = stock.Name, + Type = stock.Type, + InstrumentCategory = stock.InstrumentCategory, + HasCfd = stock.HasCfd, + ImageId = stock.ImageId, + DerivativeProductCategories = stock.DerivativeProductCategories?.ToList() ?? new List() + }, + TradeRepublicCrypto crypto => new CryptoEntity + { + Isin = crypto.Isin, + Name = crypto.Name, + Type = crypto.Type, + InstrumentCategory = crypto.InstrumentCategory, + HasCfd = crypto.HasCfd, + ImageId = crypto.ImageId + }, + TradeRepublicEtf etf => new EtfEntity + { + Isin = etf.Isin, + Name = etf.Name, + Type = etf.Type, + InstrumentCategory = etf.InstrumentCategory, + HasCfd = etf.HasCfd, + ImageId = etf.ImageId, + DerivativeProductCategories = etf.DerivativeProductCategories?.ToList() ?? new List() + }, + TradeRepublicSynthetic syn => new SyntheticEntity + { + Isin = syn.Isin, + Name = syn.Name, + Type = syn.Type, + InstrumentCategory = syn.InstrumentCategory, + HasCfd = syn.HasCfd, + ImageId = syn.ImageId, + DerivativeProductCategories = syn.DerivativeProductCategories?.ToList() ?? new List() + }, + TradeRepublicBond bond => new BondEntity + { + Isin = bond.Isin, + Name = bond.Name, + Type = bond.Type, + InstrumentCategory = bond.InstrumentCategory, + HasCfd = bond.HasCfd, + ImageId = bond.ImageId, + BondIssuerName = bond.BondIssuerName, + SearchSubtitle = bond.SearchSubtitle + }, + TradeRepublicDerivative deriv => new DerivativeEntity + { + Isin = deriv.Isin, + Name = deriv.Name, + Type = deriv.Type, + InstrumentCategory = deriv.InstrumentCategory, + HasCfd = deriv.HasCfd, + ImageId = deriv.ImageId, + UnderlyingIsin = deriv.UnderlyingIsin, + DerivativeProductCategories = deriv.DerivativeProductCategories?.ToList() ?? new List() + }, + _ => new StockEntity + { + Isin = dto.Isin, + Name = dto.Name, + Type = dto.Type, + InstrumentCategory = dto.InstrumentCategory, + HasCfd = dto.HasCfd, + ImageId = dto.ImageId } - - return await _context.TradeRepublicAssets - .Include(a => a.Tags) - .Where(a => a.Isin == isin) - .ToListAsync(); - } - - return []; + }; } - /// Inherits documentation from interface. - public async Task> GetValidAssetsByIsinAsync(string isin) + private static void UpdateSubtypeProperties(AssetEntity entity, TradeRepublicAsset dto) { - var cutoff = DateTime.UtcNow.AddDays(-14); - - return await _context.TradeRepublicAssets - .AsNoTracking() - .Include(a => a.Tags) - .Where(a => a.Isin == isin && a.LastUpdatedAt >= cutoff) - .ToListAsync(); + switch (entity) + { + case StockEntity stock when dto is TradeRepublicStock s: + stock.DerivativeProductCategories = s.DerivativeProductCategories?.ToList() ?? new List(); + break; + case EtfEntity etf when dto is TradeRepublicEtf e: + etf.DerivativeProductCategories = e.DerivativeProductCategories?.ToList() ?? new List(); + break; + case SyntheticEntity syn when dto is TradeRepublicSynthetic synDto: + syn.DerivativeProductCategories = synDto.DerivativeProductCategories?.ToList() ?? new List(); + break; + case BondEntity bond when dto is TradeRepublicBond b: + bond.BondIssuerName = b.BondIssuerName; + bond.SearchSubtitle = b.SearchSubtitle; + break; + case DerivativeEntity deriv when dto is TradeRepublicDerivative d: + deriv.UnderlyingIsin = d.UnderlyingIsin; + deriv.DerivativeProductCategories = d.DerivativeProductCategories?.ToList() ?? new List(); + break; + } } /// Inherits documentation from interface. public async Task> FindAffectedActiveAssetsAsync(string searchQuery) { - if (string.IsNullOrWhiteSpace(searchQuery)) return []; + var cutoff = DateTime.UtcNow.AddDays(-90); + string cleanQuery = searchQuery.Trim().ToLowerInvariant(); - var cutoff = DateTime.UtcNow.AddDays(-14); - - var searchTerms = searchQuery - .Split(',') - .Select(t => t.Trim()) - .Where(t => !string.IsNullOrEmpty(t)) - .Distinct() - .ToList(); - - if (searchTerms.Count == 0) return []; - - var query = _context.TradeRepublicAssets + return await _context.TradeRepublicAssets .AsNoTracking() - .Where(a => a.LastUpdatedAt >= cutoff) .Include(a => a.Tags) - .AsQueryable(); - - foreach (var term in searchTerms) - { - var lowerTerm = term.ToLower(); - query = query.Where(a => - a.Isin.ToLower().Contains(lowerTerm) || - a.Name.ToLower().Contains(lowerTerm) || - a.Tags.Any(tag => tag.Name.ToLower().Contains(lowerTerm))); - } - - var localAssets = await query.ToListAsync(); - - var possibleIsins = searchTerms - .Where(t => t.Length == 12 && char.IsLetter(t[0]) && char.IsLetter(t[1])) - .Select(t => t.ToUpper()) - .ToList(); - - if (possibleIsins.Count > 0) - { - var foundIsins = localAssets.Select(a => a.Isin).ToHashSet(); - var missingIsins = possibleIsins.Where(isin => !foundIsins.Contains(isin)).ToList(); - - if (missingIsins.Count > 0) - { - var fetchedNewAsset = false; - foreach (var missingIsin in missingIsins) - { - var trAssetDto = await _tradeRepublicService.GetAsset(missingIsin); - if (trAssetDto?.Results != null) - { - foreach (var asset in trAssetDto.Results) - { - await AddOrUpdateAssetAsync(asset); - } - - fetchedNewAsset = true; - } - } - - if (fetchedNewAsset) - { - return await query.ToListAsync(); - } - } - } - - return localAssets; + .Where(a => a.LastUpdatedAt >= cutoff && ( + a.Isin.ToLower().Contains(cleanQuery) || + a.Name.ToLower().Contains(cleanQuery) + )) + .Take(25) + .ToListAsync(); } /// Inherits documentation from interface. @@ -373,7 +376,7 @@ public class AssetsDbService : IAssetsDbService asset.ImageId = imageId; } await _context.SaveChangesAsync(); - _logger.LogInformation("[{Channel}] Updated ImageId for ISIN {Isin} in database to '{ImageId}'", "AssetsChannel", isin, imageId); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsDbService] Updated ImageId for ISIN {Isin} in database to '{ImageId}'", isin, imageId); } } @@ -383,13 +386,13 @@ public class AssetsDbService : IAssetsDbService var asset = await _context.TradeRepublicAssets.FirstOrDefaultAsync(a => a.Isin == isin); if (asset == null) { - _logger.LogWarning("[{Channel}] Delete execution cancelled. Asset with ISIN {Isin} does not exist.", "AssetsChannel", isin); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, "[AssetsDbService] Delete execution cancelled. Asset with ISIN {Isin} does not exist.", isin); return false; } _context.TradeRepublicAssets.Remove(asset); await _context.SaveChangesAsync(); - _logger.LogInformation("[{Channel}] Asset with ISIN {Isin} has been successfully deleted.", "AssetsChannel", isin); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsDbService] Asset with ISIN {Isin} has been successfully deleted.", isin); return true; } @@ -410,11 +413,10 @@ public class AssetsDbService : IAssetsDbService decimal levQuery = targetLeverage.HasValue && targetLeverage.Value > 0 ? targetLeverage.Value : 0m; - // Trade Republic uses page index (0, 1, 2, 3...) for the 'after' pagination parameter in derivatives string trAfter = !string.IsNullOrEmpty(after) ? after : (pageIndex > 0 ? pageIndex.ToString() : "0"); - _logger.LogInformation("[{Channel}] Fetching derivatives for {Isin} (OptionType: {Option}, Leverage: {Lev}, Page: {Page}, TR-After: {After})", - "AssetsChannel", underlyingIsin, cleanOptionType, levQuery, pageIndex, trAfter); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsDbService] Fetching derivatives for {Isin} (OptionType: {Option}, Leverage: {Lev}, Page: {Page}, TR-After: {After})", + underlyingIsin, cleanOptionType, levQuery, pageIndex, trAfter); var trReq = new TradeRepublicDerivativesRequest( Underlying: underlyingIsin, @@ -429,8 +431,8 @@ public class AssetsDbService : IAssetsDbService var trResponse = await _tradeRepublicService.GetDerivativesAsync(trReq, cancellationToken); var fetchedItems = trResponse?.Results ?? new List(); - _logger.LogInformation("[{Channel}] TR returned {Count} derivatives for {Isin} (Cursors.After: {NextAfter})", - "AssetsChannel", fetchedItems.Count, underlyingIsin, trResponse?.Cursors?.After ?? "null"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsDbService] TR returned {Count} derivatives for {Isin} (Cursors.After: {NextAfter})", + fetchedItems.Count, underlyingIsin, trResponse?.Cursors?.After ?? "null"); if (fetchedItems.Count > 0) { @@ -445,144 +447,75 @@ public class AssetsDbService : IAssetsDbService foreach (var item in fetchedItems) { - if (!existingDerivatives.TryGetValue(item.Isin, out var entity)) + DateTime? expiryDate = null; + if (!string.IsNullOrWhiteSpace(item.Expiry) && DateTime.TryParse(item.Expiry, out var parsedExp)) { - entity = new DerivativeEntity - { - Isin = item.Isin, - InstrumentCategory = "derivative", - Type = "derivative" - }; - await _context.TradeRepublicAssets.AddAsync(entity, cancellationToken); + expiryDate = parsedExp.ToUniversalTime(); } - bool isShortItem = string.Equals(item.OptionType, "short", StringComparison.OrdinalIgnoreCase) || - string.Equals(item.OptionType, "put", StringComparison.OrdinalIgnoreCase) || - item.OptionType.Contains("short", StringComparison.OrdinalIgnoreCase) || - item.OptionType.Contains("put", StringComparison.OrdinalIgnoreCase) || - item.OptionType.Contains("bear", StringComparison.OrdinalIgnoreCase); + if (existingDerivatives.TryGetValue(item.Isin, out var existing)) + { + existing.Name = !string.IsNullOrWhiteSpace(item.ProductCategoryName) ? item.ProductCategoryName : item.Isin; + existing.UnderlyingIsin = underlyingIsin; + existing.Strike = item.Strike ?? 0m; + existing.Barrier = item.Barrier ?? 0m; + existing.Leverage = item.Leverage ?? 0m; + existing.Expiry = expiryDate; + existing.OptionType = targetOptionType; + existing.ProductCategoryName = item.ProductCategoryName; + existing.NextGenProductCategoryName = item.NextGenProductCategoryName; + existing.Issuer = item.Issuer; + existing.IssuerDisplayName = item.IssuerDisplayName; + existing.IssuerImageId = item.IssuerImageId; + existing.Size = item.Size; + existing.Factor = item.Factor; + existing.Delta = item.Delta; + existing.Currency = item.Currency; + existing.LastUpdatedAt = now; - entity.UnderlyingIsin = underlyingIsin; - entity.OptionType = isShortItem ? OptionType.Short : OptionType.Long; - entity.ProductCategoryName = item.ProductCategoryName; - entity.NextGenProductCategoryName = item.NextGenProductCategoryName; - entity.Strike = item.Strike ?? 0m; - entity.Barrier = item.Barrier ?? 0m; - entity.Leverage = item.Leverage ?? 0m; - entity.Size = item.Size; - entity.Factor = item.Factor; - entity.Delta = item.Delta; - entity.Currency = item.Currency ?? "EUR"; - entity.Expiry = DateTime.TryParse(item.Expiry, System.Globalization.CultureInfo.InvariantCulture, System.Globalization.DateTimeStyles.AdjustToUniversal | System.Globalization.DateTimeStyles.AssumeUniversal, out var exp) - ? DateTime.SpecifyKind(exp, DateTimeKind.Utc) - : (DateTime?)null; - entity.Issuer = item.Issuer; - entity.IssuerDisplayName = item.IssuerDisplayName; - entity.IssuerImageId = item.IssuerImageId; - entity.ImageId = item.ImageId; - entity.Name = $"{item.IssuerDisplayName} {item.NextGenProductCategoryName} ({(isShortItem ? "SHORT" : "LONG")})"; - entity.LastUpdatedAt = now; + _context.TradeRepublicAssets.Update(existing); + resultEntities.Add(existing); + } + else + { + var newDeriv = new DerivativeEntity + { + Isin = item.Isin, + Name = !string.IsNullOrWhiteSpace(item.ProductCategoryName) ? item.ProductCategoryName : item.Isin, + Type = "derivative", + InstrumentCategory = "derivative", + UnderlyingIsin = underlyingIsin, + Strike = item.Strike ?? 0m, + Barrier = item.Barrier ?? 0m, + Leverage = item.Leverage ?? 0m, + Expiry = expiryDate, + OptionType = targetOptionType, + ProductCategoryName = item.ProductCategoryName, + NextGenProductCategoryName = item.NextGenProductCategoryName, + Issuer = item.Issuer, + IssuerDisplayName = item.IssuerDisplayName, + IssuerImageId = item.IssuerImageId, + Size = item.Size, + Factor = item.Factor, + Delta = item.Delta, + Currency = item.Currency, + LastUpdatedAt = now + }; - resultEntities.Add(entity); + await _context.TradeRepublicAssets.AddAsync(newDeriv, cancellationToken); + resultEntities.Add(newDeriv); + } } await _context.SaveChangesAsync(cancellationToken); return resultEntities; } - // Fallback: Query from DB if Trade Republic returned 0 or was unreachable - var dbQuery = _context.TradeRepublicAssets + return await _context.TradeRepublicAssets .OfType() .AsNoTracking() - .Include(a => a.Tags) - .Where(d => d.UnderlyingIsin == underlyingIsin && d.OptionType == targetOptionType); - - if (levQuery > 0) - { - dbQuery = dbQuery.Where(d => d.Leverage >= (levQuery - 0.2m)); - } - - var results = await dbQuery - .OrderBy(d => d.Leverage) - .Skip(pageIndex * pageSize) + .Where(d => d.UnderlyingIsin == underlyingIsin) .Take(pageSize) .ToListAsync(cancellationToken); - - return results; } - - #region Helper & Mapping Methods - - private AssetEntity MapDtoToEntity(TradeRepublicAsset dto) - { - AssetEntity entity = dto switch - { - TradeRepublicStock stock => new StockEntity - { Isin = stock.Isin, DerivativeProductCategories = stock.DerivativeProductCategories.ToList() }, - TradeRepublicCrypto crypto => new CryptoEntity - { Isin = crypto.Isin, Subtitle = crypto.Subtitle, SearchSubtitle = crypto.SearchSubtitle }, - TradeRepublicEtf etf => new EtfEntity - { - Isin = etf.Isin, EtfDescription = etf.EtfDescription, MappedEtfIndexName = etf.MappedEtfIndexName, - Subtitle = etf.Subtitle, SearchSubtitle = etf.SearchSubtitle, - DerivativeProductCategories = etf.DerivativeProductCategories.ToList() - }, - TradeRepublicSynthetic synth => new SyntheticEntity - { Isin = synth.Isin, DerivativeProductCategories = synth.DerivativeProductCategories.ToList() }, - TradeRepublicBond bond => new BondEntity - { Isin = bond.Isin, BondIssuerName = bond.BondIssuerName, SearchSubtitle = bond.SearchSubtitle }, - TradeRepublicDerivative deriv => new DerivativeEntity - { - Isin = deriv.Isin, UnderlyingIsin = deriv.UnderlyingIsin, - DerivativeProductCategories = deriv.DerivativeProductCategories.ToList() - }, - _ => throw new NotSupportedException($"Type {dto.GetType().Name} is not supported.") - }; - - return PopulateBaseProperties(entity, dto); - } - - private AssetEntity PopulateBaseProperties(AssetEntity entity, TradeRepublicAsset dto) - { - entity.Name = dto.Name; - entity.Type = dto.Type; - entity.InstrumentCategory = dto.InstrumentCategory; - entity.HasCfd = dto.HasCfd; - entity.ImageId = dto.ImageId; - return entity; - } - - private void UpdateSubtypeProperties(AssetEntity entity, TradeRepublicAsset dto) - { - switch (entity) - { - case StockEntity stockEntity when dto is TradeRepublicStock stockDto: - stockEntity.DerivativeProductCategories = stockDto.DerivativeProductCategories.ToList(); - break; - case CryptoEntity cryptoEntity when dto is TradeRepublicCrypto cryptoDto: - cryptoEntity.Subtitle = cryptoDto.Subtitle; - cryptoEntity.SearchSubtitle = cryptoDto.SearchSubtitle; - break; - case EtfEntity etfEntity when dto is TradeRepublicEtf etfDto: - etfEntity.DerivativeProductCategories = etfDto.DerivativeProductCategories.ToList(); - etfEntity.EtfDescription = etfDto.EtfDescription; - etfEntity.MappedEtfIndexName = etfDto.MappedEtfIndexName; - etfEntity.Subtitle = etfDto.Subtitle; - etfEntity.SearchSubtitle = etfDto.SearchSubtitle; - break; - case SyntheticEntity synthEntity when dto is TradeRepublicSynthetic synthDto: - synthEntity.DerivativeProductCategories = synthDto.DerivativeProductCategories.ToList(); - break; - case BondEntity bondEntity when dto is TradeRepublicBond bondDto: - bondEntity.BondIssuerName = bondDto.BondIssuerName; - bondEntity.SearchSubtitle = bondDto.SearchSubtitle; - break; - case DerivativeEntity derivEntity when dto is TradeRepublicDerivative derivDto: - derivEntity.DerivativeProductCategories = derivDto.DerivativeProductCategories.ToList(); - derivEntity.UnderlyingIsin = derivDto.UnderlyingIsin; - break; - } - } - - #endregion } \ No newline at end of file diff --git a/FinlyticAssets/Services/AssetsIndexService.cs b/FinlyticAssets/Services/AssetsIndexService.cs index 4955462..ffebb85 100644 --- a/FinlyticAssets/Services/AssetsIndexService.cs +++ b/FinlyticAssets/Services/AssetsIndexService.cs @@ -1,7 +1,13 @@ +using System; +using System.IO; +using System.Linq; +using System.Net.Http; using System.Text.Json; -using System.Text.Json.Serialization; +using System.Threading; +using System.Threading.Tasks; using FinlyticAssets.Models; using FinlyticAssets.Util; +using FinlyticCore.Services; namespace FinlyticAssets.Services; @@ -13,8 +19,6 @@ public interface IAssetsIndexService /// /// Recreates the index file containing basic asset identifiers (ISIN and Name) for all active, valid assets. /// - /// A token to monitor for cancellation requests. - /// A task that represents the asynchronous operation. public Task ReCreateIndexFileAsync(CancellationToken cancellationToken = default); /// @@ -28,18 +32,13 @@ public interface IAssetsIndexService /// public class AssetsIndexService : IAssetsIndexService { - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; private readonly IAssetsDbService _assetsDbService; private static readonly HttpClient _httpClient = new(); - /// - /// Initializes a new instance of the class. - /// - /// The logger for documenting indexing events and errors. - /// The database service to query the assets from. - public AssetsIndexService(ILogger logger, IAssetsDbService assetsDbService) + public AssetsIndexService(IFinlyticLogger finlyticLogger, IAssetsDbService assetsDbService) { - _logger = logger; + _finlyticLogger = finlyticLogger; _assetsDbService = assetsDbService; } @@ -51,7 +50,7 @@ public class AssetsIndexService : IAssetsIndexService var assets = await _assetsDbService.GetAllValidAssetsAsync(); if (assets == null || !assets.Any()) { - _logger.LogWarning("[{Channel}] No valid assets found in the database to index.", "AssetsChannel"); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, "[AssetsIndexService] No valid assets found in the database to index."); return; } @@ -59,7 +58,6 @@ public class AssetsIndexService : IAssetsIndexService .DistinctBy(a => a.Isin) .Select(a => { string cleanIsin = a.Isin.Trim().ToUpperInvariant(); - // Point directly to our own local backend logo endpoint string imageUrl = $"/api/v1/logo/{cleanIsin}"; return new AssetIndex(cleanIsin, a.Name, imageUrl); }).ToList(); @@ -78,22 +76,22 @@ public class AssetsIndexService : IAssetsIndexService await JsonSerializer.SerializeAsync(fileStream, indexAssets, cancellationToken: cancellationToken); } - _logger.LogInformation("[{Channel}] Successfully recreated asset index file with {Count} entries pointing to local logos at {Path}", - "AssetsChannel", indexAssets.Count, filePath); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsIndexService] Successfully recreated asset index file with {Count} entries pointing to local logos at {Path}", + indexAssets.Count, filePath); } catch (IOException ex) { - _logger.LogError(ex, "[{Channel}] Disk I/O error occurred while writing the asset index file.", "AssetsChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Disk I/O error occurred while writing the asset index file."); throw; } catch (JsonException ex) { - _logger.LogError(ex, "[{Channel}] Failed to serialize the asset index data to JSON.", "AssetsChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Failed to serialize the asset index data to JSON."); throw; } catch (Exception ex) { - _logger.LogError(ex, "[{Channel}] An unexpected error occurred while recreating the asset index file.", "AssetsChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] An unexpected error occurred while recreating the asset index file."); throw; } } @@ -131,13 +129,13 @@ public class AssetsIndexService : IAssetsIndexService { byte[] data = await response.Content.ReadAsByteArrayAsync(cancellationToken); await File.WriteAllBytesAsync(filePath, data, cancellationToken); - _logger.LogInformation("[{Channel}] Successfully saved logo SVG for ISIN {Isin} to {Path} on demand", "AssetsChannel", cleanIsin, filePath); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[AssetsIndexService] Successfully saved logo SVG for ISIN {Isin} to {Path} on demand", cleanIsin, filePath); return filePath; } } catch (Exception ex) { - _logger.LogWarning(ex, "[{Channel}] Failed to download logo for ISIN {Isin} from {Url}", "AssetsChannel", cleanIsin, targetUrl); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, ex, "[AssetsIndexService] Failed to download logo for ISIN {Isin} from {Url}", cleanIsin, targetUrl); } return null; diff --git a/FinlyticAssets/Services/LogoFetcherBackgroundService.cs b/FinlyticAssets/Services/LogoFetcherBackgroundService.cs index 76b1c1f..2530a2d 100644 --- a/FinlyticAssets/Services/LogoFetcherBackgroundService.cs +++ b/FinlyticAssets/Services/LogoFetcherBackgroundService.cs @@ -6,9 +6,9 @@ using System.Text; using System.Threading; using System.Threading.Tasks; using FinlyticAssets.Util; +using FinlyticCore.Services; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; -using Microsoft.Extensions.Logging; namespace FinlyticAssets.Services; @@ -19,7 +19,7 @@ namespace FinlyticAssets.Services; /// public class LogoFetcherBackgroundService : BackgroundService { - private readonly ILogger _logger; + private readonly IFinlyticLogger _finlyticLogger; private readonly IServiceScopeFactory _scopeFactory; private readonly HttpClient _httpClient; @@ -32,10 +32,10 @@ public class LogoFetcherBackgroundService : BackgroundService """; public LogoFetcherBackgroundService( - ILogger logger, + IFinlyticLogger finlyticLogger, IServiceScopeFactory scopeFactory) { - _logger = logger; + _finlyticLogger = finlyticLogger; _scopeFactory = scopeFactory; _httpClient = new HttpClient(); _httpClient.DefaultRequestHeaders.TryAddWithoutValidation("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"); @@ -45,7 +45,7 @@ public class LogoFetcherBackgroundService : BackgroundService protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - _logger.LogInformation("[{Channel}] LogoFetcherBackgroundService started. Will fetch missing logos periodically.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] LogoFetcherBackgroundService started. Will fetch missing logos periodically."); await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken); @@ -57,7 +57,7 @@ public class LogoFetcherBackgroundService : BackgroundService } catch (Exception ex) when (!stoppingToken.IsCancellationRequested) { - _logger.LogError(ex, "[{Channel}] Error occurred while executing logo batch fetch.", "AssetsChannel"); + await _finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[LogoFetcherBackgroundService] Error occurred while executing logo batch fetch."); } try @@ -70,7 +70,7 @@ public class LogoFetcherBackgroundService : BackgroundService } } - _logger.LogInformation("[{Channel}] LogoFetcherBackgroundService stopped.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] LogoFetcherBackgroundService stopped."); } private async Task ProcessMissingLogosBatchAsync(CancellationToken stoppingToken) @@ -88,7 +88,6 @@ public class LogoFetcherBackgroundService : BackgroundService Directory.CreateDirectory(directoryPath); } - // ✅ Prüft sowohl DB-Eintrag ALS AUCH, ob die Datei bereits lokal existiert var missingIsins = validAssets .Select(a => a.Isin?.Trim().ToUpperInvariant()) .Where(isin => !string.IsNullOrEmpty(isin)) @@ -98,12 +97,12 @@ public class LogoFetcherBackgroundService : BackgroundService if (missingIsins.Count == 0) { - _logger.LogDebug("All asset logos are downloaded and up to date."); + await _finlyticLogger.LogDebugAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] All asset logos are downloaded and up to date."); return; } var batchToFetch = missingIsins.Take(60).ToList(); - _logger.LogInformation("[{Channel}] Found {Count} missing logos on disk. Fetching bulk batch of {BatchSize} logos...", "AssetsChannel", missingIsins.Count, batchToFetch.Count); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] Found {Count} missing logos on disk. Fetching bulk batch of {BatchSize} logos...", missingIsins.Count, batchToFetch.Count); int successCount = 0; @@ -126,7 +125,7 @@ public class LogoFetcherBackgroundService : BackgroundService } else { - _logger.LogWarning("[{Channel}] Logo not found on CDN for ISIN {Isin} (HTTP {StatusCode}). Saving SVG placeholder.", "AssetsChannel", isin, response.StatusCode); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] Logo not found on CDN for ISIN {Isin} (HTTP {StatusCode}). Saving SVG placeholder.", isin, response.StatusCode); byte[] placeholderData = Encoding.UTF8.GetBytes(PlaceholderSvg); await File.WriteAllBytesAsync(filePath, placeholderData, stoppingToken); successCount++; @@ -136,7 +135,7 @@ public class LogoFetcherBackgroundService : BackgroundService } catch (Exception ex) when (!stoppingToken.IsCancellationRequested) { - _logger.LogWarning(ex, "[{Channel}] Exception while downloading logo for ISIN {Isin} from {Url}. Saving SVG placeholder.", "AssetsChannel", isin, targetUrl); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, ex, "[LogoFetcherBackgroundService] Exception while downloading logo for ISIN {Isin} from {Url}. Saving SVG placeholder.", isin, targetUrl); try { byte[] placeholderData = Encoding.UTF8.GetBytes(PlaceholderSvg); @@ -148,23 +147,22 @@ public class LogoFetcherBackgroundService : BackgroundService catch { } } - // Kurze Pause gegen Rate Limiting await Task.Delay(50, stoppingToken); } - _logger.LogInformation("[{Channel}] Batch fetch complete. Successfully processed {SuccessCount}/{BatchSize} logos. Remaining missing: {Remaining}", - "AssetsChannel", successCount, batchToFetch.Count, missingIsins.Count - batchToFetch.Count); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] Batch fetch complete. Successfully processed {SuccessCount}/{BatchSize} logos. Remaining missing: {Remaining}", + successCount, batchToFetch.Count, missingIsins.Count - batchToFetch.Count); if (successCount > 0) { try { await indexService.ReCreateIndexFileAsync(stoppingToken); - _logger.LogInformation("[{Channel}] Successfully updated index.json after logo batch fetch.", "AssetsChannel"); + await _finlyticLogger.LogInfoAsync(SettingKeys.AssetsChannel, "[LogoFetcherBackgroundService] Successfully updated index.json after logo batch fetch."); } catch (Exception ex) { - _logger.LogWarning(ex, "[{Channel}] Failed to update index.json after logo batch fetch.", "AssetsChannel"); + await _finlyticLogger.LogWarningAsync(SettingKeys.AssetsChannel, ex, "[LogoFetcherBackgroundService] Failed to update index.json after logo batch fetch."); } } } diff --git a/FinlyticAssets/Util/AssetsMqttClient.cs b/FinlyticAssets/Util/AssetsMqttClient.cs index 3532b02..6dc303c 100644 --- a/FinlyticAssets/Util/AssetsMqttClient.cs +++ b/FinlyticAssets/Util/AssetsMqttClient.cs @@ -1,75 +1,94 @@ +using System; +using System.Collections.Generic; +using System.Linq; using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; using FinlyticAssets.Entities; using FinlyticAssets.Services; +using FinlyticCore.Dtos.Settings; using FinlyticCore.Models; +using FinlyticCore.Services; using FinlyticCore.Util; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; namespace FinlyticAssets.Util; /// /// 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 and index service, -/// and publishes the requested asset entities or logo files back to the response topic. -/// Also implements to manage its own lifecycle connections. /// -public class AssetsMqttClient( - ILogger logger, - IServiceScopeFactory scopeFactory, - IConfiguration configuration) : ManagedMqttClient(logger), IHostedService +public class AssetsMqttClient : ManagedMqttClient, IHostedService { + private readonly ILogger _logger; + private readonly IServiceScopeFactory _scopeFactory; + private readonly IConfiguration _configuration; + public AssetsMqttClient( + ILogger logger, + IServiceScopeFactory scopeFactory, + IConfiguration configuration) : base(logger) + { + _logger = logger; + _scopeFactory = scopeFactory; + _configuration = configuration; + } /// /// Starts the MQTT client and connects to the configured broker. /// - /// A token to monitor for cancellation requests. - /// A task representing the asynchronous start operation. public async Task StartAsync(CancellationToken cancellationToken) { 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()}" + Host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost", + Port = Convert.ToInt32(_configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883"), + ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "FinlyticAssets")}_{Guid.NewGuid()}" }; + _logger.LogInformation("Starting Assets MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId); await ConnectAsync(config); } /// /// Gracefully stops and disconnects the MQTT client. /// - /// A token to monitor for cancellation requests. - /// A task representing the asynchronous stop operation. public async Task StopAsync(CancellationToken cancellationToken) { + _logger.LogInformation("Stopping Assets MQTT client."); await DisconnectAsync(); } /// /// Invoked automatically once the connection to the MQTT broker is successfully established or restored. - /// Registers subscriptions for asset validation, search, and logo download requests. /// - /// A representing the asynchronous subscription operation. protected override async Task OnConnectedAsync() { + _logger.LogInformation("Assets MQTT Client connected. Subscribing to topics..."); await SubscribeAsync("services/request/assets_Get/#"); await SubscribeAsync("services/request/assets_Search/#"); await SubscribeAsync("services/request/assets_GetDiscovery/#"); await SubscribeAsync("services/request/assets_GetDerivatives/#"); await SubscribeAsync("services/request/assets_FetchLogo/#"); + await SubscribeAsync("services/request/assets_settings_GetAll/#"); + await SubscribeAsync("services/request/assets_settings_Update/#"); await SubscribeAsync("services/request/health_Ping/#"); await SubscribeAsync("services/config/updated/#"); + + FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) => + { + if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticAssets", StringComparison.OrdinalIgnoreCase)) + { + await PublishAsync("finlytic/logs/FinlyticAssets", logDto); + } + }; } - /// - /// 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. + /// Processes incoming messages on the subscribed topics. /// - /// The MQTT topic on which the message was received. - /// The incoming message as a UTF-8 encoded JSON string. - /// A representing the asynchronous message processing operation. protected override async Task OnMessageReceivedAsync(string topic, string payload) { if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase)) @@ -90,9 +109,21 @@ public class AssetsMqttClient( return; } + if (topic.StartsWith("services/request/assets_settings_GetAll", StringComparison.OrdinalIgnoreCase)) + { + await HandleSettingsGetAllAsync(correlationId); + return; + } + + if (topic.StartsWith("services/request/assets_settings_Update", StringComparison.OrdinalIgnoreCase)) + { + await HandleSettingsUpdateAsync(payload, correlationId); + return; + } + try { - using var scope = scopeFactory.CreateScope(); + using var scope = _scopeFactory.CreateScope(); var dbService = scope.ServiceProvider.GetRequiredService(); var indexService = scope.ServiceProvider.GetRequiredService(); @@ -129,26 +160,90 @@ public class AssetsMqttClient( } } + private async Task HandleSettingsGetAllAsync(string correlationId) + { + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + var settingsService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId); + try + { + var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); + var responseTopic = $"services/response/assets_settings_GetAll/{correlationId}"; + + await PublishAsync(responseTopic, settings); + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic); + } + catch (Exception ex) + { + await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAssets] [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>(); + var settingsService = scope.ServiceProvider.GetRequiredService(); + + await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticAssets] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId); + try + { + Dictionary? updates = null; + try + { + updates = JsonSerializer.Deserialize>(payload); + } + catch + { + var list = JsonSerializer.Deserialize>(payload); + if (list != null) + { + updates = new Dictionary(); + 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, "[FinlyticAssets] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count); + } + + var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) }); + var responseTopic = $"services/response/assets_settings_Update/{correlationId}"; + await PublishAsync(responseTopic, currentSettings); + } + catch (Exception ex) + { + await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticAssets] [Settings_Update] Failed to update settings."); + } + } + 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(); - await settingsDb.UpdateSettingsFromDictionary(updatePayload.Settings); - logger.LogInformation("[{Channel}] [AssetsMqttClient] Persisted {Count} updated settings to FinlyticAssets database.", "AssetsChannel", updatePayload.Settings.Count); + using var scope = _scopeFactory.CreateScope(); + var settings = scope.ServiceProvider.GetRequiredService(); + var dict = updatePayload.Settings.ToDictionary(k => k.Key, v => (object?)v.Value); + await settings.UpdateSettingsAsync(dict); } } catch (Exception ex) { - logger.LogError(ex, "[{Channel}] [AssetsMqttClient] Error processing MQTT config update event.", "AssetsChannel"); + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + await finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsMqttClient] Error processing MQTT config update event."); } } @@ -162,7 +257,9 @@ public class AssetsMqttClient( { 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); + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[AssetsMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId); } } @@ -236,7 +333,9 @@ public class AssetsMqttClient( } catch (Exception ex) { - logger.LogError(ex, "[{Channel}] Error parsing GetDerivativesRequest payload.", "AssetsChannel"); + using var scope = _scopeFactory.CreateScope(); + var finlyticLogger = scope.ServiceProvider.GetRequiredService>(); + await finlyticLogger.LogErrorAsync(SettingKeys.AssetsChannel, ex, "[AssetsMqttClient] Error parsing GetDerivativesRequest payload."); } return []; diff --git a/FinlyticAssets/Util/SettingKeys.cs b/FinlyticAssets/Util/SettingKeys.cs new file mode 100644 index 0000000..e0794be --- /dev/null +++ b/FinlyticAssets/Util/SettingKeys.cs @@ -0,0 +1,22 @@ +using FinlyticCore.Models.Settings; + +namespace FinlyticAssets.Util; + +public static class SettingKeys +{ + // --- Logging-Kanäle --- + public static readonly SettingKey AssetsChannel = new("Logging.Channel.Assets", true); + public static readonly SettingKey MqttChannel = new("Logging.Channel.MQTT", true); + public static readonly SettingKey HealthPingChannel = new("Logging.Channel.Health", true); + + // --- Asset Scanning --- + public static readonly SettingKey EnableAutoScan = new("Scanner.EnableAutoScan", true); + public static readonly SettingKey ScanIntervalHours = new("Scanner.ScanIntervalHours", 12); + public static readonly SettingKey MaxConcurrentScans = new("Scanner.MaxConcurrentScans", 5); + public static readonly SettingKey EnableDerivativeScanning = new("Scanner.EnableDerivativeScanning", true); + + // --- Logos & Media --- + public static readonly SettingKey AutoFetchLogos = new("Media.AutoFetchLogos", true); + public static readonly SettingKey LogoFetchBatchSize = new("Media.LogoFetchBatchSize", 25); + public static readonly SettingKey LogoStorageDirectory = new("Media.LogoStorageDirectory", "data/logos"); +}