feat(fundamentals): add KeyedLockPool for concurrent scraping synchronization and update MQTT RPC handlers

FinlyticFundamentals/Program.cs

@@ -18,8 +18,7 @@ var builder = Host.CreateApplicationBuilder(args);
 
 // Register DB Context & ISettingsDbContext
 builder.Services.AddDbContext<FundamentalsDbContext>(options =>
-    options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection"))
+    options.UseNpgsql(builder.Configuration.GetConnectionString("DefaultConnection")));
-           .ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning)));
 builder.Services.AddScoped<ISettingsDbContext>(sp => sp.GetRequiredService<FundamentalsDbContext>());
 
 // Register HTTP Clients
@@ -57,8 +56,10 @@ using (var scope = host.Services.CreateScope())
     try
     {
         var context = scope.ServiceProvider.GetRequiredService<FundamentalsDbContext>();
-        await context.Database.MigrateAsync();
+        var connStr = builder.Configuration.GetConnectionString("DefaultConnection") ?? "";
+        await context.MigrateWithBootstrapAsync(connStr);
         Console.WriteLine("Database migrations successfully executed for FinlyticFundamentals.");
+
     }
     catch (Exception ex)
     {

FinlyticFundamentals/Project.md

@@ -1,31 +0,0 @@
-# Finlytic Fundamentals Service
-
-Finlytic Fundamentals is a C# background worker microservice responsible for fetching, caching, and serving financial fundamental data (P/E ratios, market cap, dividend yield, revenue growth, corporate calendar events) across global equities.
-
----
-
-## Core Features & Architecture
-
-1. **Fundamental Data Ingestion**:
-   - Scrapes and ingests company fundamentals (`AssetFundamentalsDto`) including P/E, EPS, Market Cap, Dividend Yield, Revenue, Profit Margins, and Debt-to-Equity ratios.
-
-2. **Corporate Event Calendar**:
-   - Tracks earnings release dates, ex-dividend dates, payout dates, and shareholder meetings (`CorporateEventDto`).
-
-3. **MQTT Distribution Channels**:
-   - Publishes fundamental updates to `finlytic/fundamentals/{isin}` and `finlytic/assets/fundamentals/{isin}`.
-   - Responds to RPC requests on `services/request/fundamentals_Get/#` and `services/request/events_GetAll/#`.
-
----
-
-## Feature Status
-
-### Implemented Features
-- [x] Fundamentals Database Persistence & Caching (`FundamentalsDbContext`).
-- [x] Corporate Event Calendar storage & query handlers.
-- [x] Zero-Allocation MQTT serialization via `FinlyticJsonSerializerContext`.
-- [x] Pure Worker Service architecture (`Host.CreateApplicationBuilder`, no Kestrel HTTP server).
-
-### Planned Features
-- [ ] Financial Modeling Prep / SEC EDGAR API automated quarterly filing sync.
-- [ ] Automated Dividend Growth Rate & Dividend Safety Rating calculation engine.

FinlyticFundamentals/Services/FundamentalsDbService.cs

@@ -35,7 +35,7 @@ public interface IFundamentalsDbService
 
 public class FundamentalsDbService : IFundamentalsDbService
 {
-    private static readonly ConcurrentDictionary<string, SemaphoreSlim> IsinLocks = new();
+    private static readonly KeyedLockPool LockPool = new();
 
     private readonly IServiceScopeFactory _scopeFactory;
     private readonly IYahooFinanceScraper _scraper;
@@ -65,15 +65,13 @@ public class FundamentalsDbService : IFundamentalsDbService
         var cleanIsin = isin.Trim().ToUpperInvariant();
         var requestedTicker = ticker?.Trim().ToUpperInvariant();
 
-        var isinLock = IsinLocks.GetOrAdd(cleanIsin, _ => new SemaphoreSlim(1, 1));
+        using (await LockPool.LockAsync(cleanIsin, cancellationToken))
-        await isinLock.WaitAsync(cancellationToken);
-
-        try
         {
             using var scope = _scopeFactory.CreateScope();
             var context = scope.ServiceProvider.GetRequiredService<FundamentalsDbContext>();
             var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
 
+
             // 1. Dynamic Settings lesen
             bool allowForceRefresh =
                 await settingsService.GetSettingAsync(SettingKeys.AllowForceRefresh, cancellationToken);
@@ -380,6 +378,46 @@ public class FundamentalsDbService : IFundamentalsDbService
                             assetData.AssetEvents.Add(newEvent);
                         }
                     }
+
+                    // Structured Trade Republic dividend data (ExpectedDividend + historical Dividends) carries
+                    // a real ExDate per entry - a far more reliable "this is a dividend" signal than matching
+                    // the generic Events/PastEvents feed's free-text Type/Title strings above, whose exact
+                    // wording for dividend entries is not guaranteed. The canonical "Dividend" Type here is
+                    // fully controlled by this codebase (not guessed from TR's free text), so
+                    // AssetFundamentalsDto.DaysToNextExDividend can match on it reliably (Rules.md §4).
+                    var trDividendList = new List<TradeRepublicDividendDto>();
+                    if (trDetails.ExpectedDividend != null) trDividendList.Add(trDetails.ExpectedDividend);
+                    if (trDetails.Dividends != null) trDividendList.AddRange(trDetails.Dividends);
+
+                    foreach (var div in trDividendList)
+                    {
+                        if (string.IsNullOrWhiteSpace(div.ExDate) ||
+                            !DateTime.TryParse(div.ExDate, System.Globalization.CultureInfo.InvariantCulture,
+                                System.Globalization.DateTimeStyles.None, out var exDate))
+                        {
+                            continue;
+                        }
+
+                        bool isDuplicateDividend = assetData.AssetEvents.Any(e =>
+                            e.Date.Date == exDate.Date && string.Equals(e.Type, "Dividend", StringComparison.OrdinalIgnoreCase));
+
+                        if (!isDuplicateDividend)
+                        {
+                            var newDividendEvent = new AssetEventEntity
+                            {
+                                AssetDataIsin = cleanIsin,
+                                Ticker = new TickerEntity
+                                {
+                                    Ticker = yahooPrimaryTicker.Ticker,
+                                    Exchange = yahooPrimaryTicker.Exchange ?? "Unknown"
+                                },
+                                Type = "Dividend",
+                                Date = exDate.Date
+                            };
+                            context.AssetEvents.Add(newDividendEvent);
+                            assetData.AssetEvents.Add(newDividendEvent);
+                        }
+                    }
                 }
 
                 // --- Process Modules DTO (Executives & Fundamental Data) ---
@@ -579,12 +617,9 @@ public class FundamentalsDbService : IFundamentalsDbService
 
             return MapToDto(assetData, fundamentalData, executivesList, eventsList);
         }
-        finally
-        {
-            isinLock.Release();
-        }
     }
 
+
     /// <inheritdoc />
     public async Task<List<CorporateEventDto>> GetAllEventsAsync(CancellationToken cancellationToken = default)
     {

FinlyticFundamentals/Services/KeyedLockPool.cs

@@ -0,0 +1,66 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace FinlyticFundamentals.Services;
+
+public class KeyedLockPool
+{
+    private readonly ConcurrentDictionary<string, RefCountedSemaphore> _semaphores = new(StringComparer.OrdinalIgnoreCase);
+
+    public async Task<IDisposable> LockAsync(string key, CancellationToken cancellationToken = default)
+    {
+        var item = _semaphores.AddOrUpdate(
+            key,
+            _ => new RefCountedSemaphore(),
+            (_, existing) =>
+            {
+                Interlocked.Increment(ref existing.RefCount);
+                return existing;
+            });
+
+        await item.Semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
+        return new Releaser(this, key, item);
+    }
+
+    private void Release(string key, RefCountedSemaphore item)
+    {
+        item.Semaphore.Release();
+        if (Interlocked.Decrement(ref item.RefCount) <= 0)
+        {
+            _semaphores.TryRemove(new KeyValuePair<string, RefCountedSemaphore>(key, item));
+        }
+    }
+
+    private sealed class RefCountedSemaphore
+    {
+        public int RefCount = 1;
+        public readonly SemaphoreSlim Semaphore = new(1, 1);
+    }
+
+    private sealed class Releaser : IDisposable
+    {
+        private readonly KeyedLockPool _pool;
+        private readonly string _key;
+        private readonly RefCountedSemaphore _item;
+        private bool _disposed;
+
+        public Releaser(KeyedLockPool pool, string key, RefCountedSemaphore item)
+        {
+            _pool = pool;
+            _key = key;
+            _item = item;
+        }
+
+        public void Dispose()
+        {
+            if (!_disposed)
+            {
+                _disposed = true;
+                _pool.Release(_key, _item);
+            }
+        }
+    }
+}

FinlyticFundamentals/Util/FundamentalsMqttClient.cs

@@ -35,12 +35,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
     /// <inheritdoc />
     public async Task StartAsync(CancellationToken cancellationToken)
     {
-        var config = new MqttConfiguration
+        var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticFundamentals");
-        {
-            Host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost",
-            Port = Convert.ToInt32(_configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883"),
-            ClientId = _configuration["MQTT:ClientId"] ?? "finlytic_fundamentals_" + Guid.NewGuid().ToString("N")
-        };
 
         _logger.LogInformation("[{Channel}] [MQTT_Client] Starting Fundamentals MQTT client. Host: {Host}, ClientId: {ClientId}", "MqttChannel", config.Host, config.ClientId);
 
@@ -58,18 +53,19 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
     protected override async Task OnConnectedAsync()
     {
         _logger.LogInformation("[{Channel}] [MQTT_Client] Connected. Subscribing to RPC request topics...", "MqttChannel");
-        await SubscribeAsync("services/request/fundamentals_Get/#");
+        await SubscribeAsync(MqttTopics.ResponseWildcard);
-        await SubscribeAsync("services/request/events_GetAll/#");
+        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsGet));
-        await SubscribeAsync("services/request/events_GetByMonth/#");
+        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.EventsGetAll));
-        await SubscribeAsync("services/request/fundamentals_settings_GetAll/#");
+        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.EventsGetByMonth));
-        await SubscribeAsync("services/request/fundamentals_settings_Update/#");
+        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsSettingsGetAll));
-        await SubscribeAsync("services/request/health_Ping/#");
+        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.FundamentalsSettingsUpdate));
+        await SubscribeAsync(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing));
 
         FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
         {
             if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticFundamentals", StringComparison.OrdinalIgnoreCase))
             {
-                await PublishAsync("finlytic/logs/FinlyticFundamentals", logDto);
+                await PublishAsync(MqttTopics.Logs("FinlyticFundamentals"), logDto);
             }
         };
     }
@@ -84,27 +80,27 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
 
         var correlationId = topic.Substring(lastSlash + 1);
 
-        if (topic.StartsWith("services/request/fundamentals_Get", StringComparison.OrdinalIgnoreCase))
+        if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsGet, StringComparison.OrdinalIgnoreCase))
         {
             await OnFundamentalsGetAsync(payload, correlationId);
         }
-        else if (topic.StartsWith("services/request/events_GetAll", StringComparison.OrdinalIgnoreCase))
+        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.EventsGetAll, StringComparison.OrdinalIgnoreCase))
         {
             await OnEventsGetAllAsync(correlationId);
         }
-        else if (topic.StartsWith("services/request/events_GetByMonth", StringComparison.OrdinalIgnoreCase))
+        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.EventsGetByMonth, StringComparison.OrdinalIgnoreCase))
         {
             await OnEventsGetByMonthAsync(payload, correlationId);
         }
-        else if (topic.StartsWith("services/request/fundamentals_settings_GetAll", StringComparison.OrdinalIgnoreCase))
+        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsSettingsGetAll, StringComparison.OrdinalIgnoreCase))
         {
             await OnSettingsGetAllAsync(correlationId);
         }
-        else if (topic.StartsWith("services/request/fundamentals_settings_Update", StringComparison.OrdinalIgnoreCase))
+        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.FundamentalsSettingsUpdate, StringComparison.OrdinalIgnoreCase))
         {
             await OnSettingsUpdateAsync(payload, correlationId);
         }
-        else if (topic.StartsWith("services/request/health_Ping", StringComparison.OrdinalIgnoreCase))
+        else if (topic.StartsWith(MqttTopics.RequestPrefix + MqttTopics.Channels.HealthPing, StringComparison.OrdinalIgnoreCase))
         {
             await OnHealthPingAsync(topic, correlationId);
         }
@@ -135,7 +131,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
                 request.Isin, request.ForceRefresh.ToString(), correlationId);
 
             var fundamentals = await dbService.GetFundamentalsAsync(request.Isin, request.Ticker, request.ForceRefresh);
-            var responseTopic = $"services/response/fundamentals_Get/{correlationId}";
+            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsGet, correlationId);
 
             await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing RPC fundamentals response to '{ResponseTopic}'", responseTopic);
             await PublishAsync(responseTopic, fundamentals);
@@ -156,7 +152,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
         try
         {
             var events = await dbService.GetAllEventsAsync();
-            var responseTopic = $"services/response/events_GetAll/{correlationId}";
+            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.EventsGetAll, correlationId);
 
             await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing events RPC response to '{ResponseTopic}'", responseTopic);
             await PublishAsync(responseTopic, events);
@@ -183,7 +179,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
             await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Processing RPC events_GetByMonth request for {Year}/{Month} [CorrelationId: {CorrelationId}]", request.Year, request.Month, correlationId);
             
             var events = await dbService.GetEventsByMonthAsync(request.Year, request.Month);
-            var responseTopic = $"services/response/events_GetByMonth/{correlationId}";
+            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.EventsGetByMonth, correlationId);
 
             await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [MQTT_Client] Publishing monthly events RPC response to '{ResponseTopic}'", responseTopic);
             await PublishAsync(responseTopic, events);
@@ -204,7 +200,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
         try
         {
             var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
-            var responseTopic = $"services/response/fundamentals_settings_GetAll/{correlationId}";
+            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsSettingsGetAll, correlationId);
 
             await PublishAsync(responseTopic, settings);
             await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticFundamentals] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
@@ -251,7 +247,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
             }
 
             var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
-            var responseTopic = $"services/response/fundamentals_settings_Update/{correlationId}";
+            var responseTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.FundamentalsSettingsUpdate, correlationId);
             await PublishAsync(responseTopic, currentSettings);
         }
         catch (Exception ex)
@@ -267,7 +263,7 @@ public class FundamentalsMqttClient : ManagedMqttClient, IHostedService
             await using var scope = _scopeFactory.CreateAsyncScope();
             var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<FundamentalsMqttClient>>();
 
-            var respTopic = $"services/response/health_Ping/{correlationId}";
+            var respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
             await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticFundamentals", "Online", DateTime.UtcNow, "Connected"));
             await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "[FinlyticFundamentals] [Health_Ping] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
         }