fix(tools.console): include snapshot oslo projection when waiting to process + add activity logging

Author: rikdepeuter

src/BuildingRegistry.Tools.Console/Infrastructure/ProjectionRepository.cs

@@ -23,9 +23,15 @@ public async Task<long> GetProducerPosition()
         {
             await using var sql = new Microsoft.Data.SqlClient.SqlConnection(_eventsConnectionString);
             return await sql.ExecuteScalarAsync<long>($@"
-                SELECT TOP 1 COALESCE([Position], 0)
-                FROM [{Schema.Producer}].[ProjectionStates]
-                ORDER BY [Position] ASC");
+SELECT TOP 1 COALESCE([Position], 0)
+FROM (
+SELECT [Position]
+FROM [{Schema.Producer}].[ProjectionStates]
+UNION
+SELECT [Position]
+FROM [{Schema.ProducerSnapshotOslo}].[ProjectionStates]
+) as q
+ORDER BY [Position] ASC");
         }
     }
 }

src/BuildingRegistry.Tools.Console/RepairBuilding/RepairBuildingService.cs

@@ -40,19 +40,24 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
             await _repairBuildingRepository.EnsureSchemaAndTablesExist();
             await _repairBuildingRepository.FillBuildingToProcess();
-            var idsToProcess = await _repairBuildingRepository.GetBuildingsToProcess();
+            var idsToProcess = (await _repairBuildingRepository.GetBuildingsToProcess()).ToList();
+            _logger.LogInformation("Found {idsToProcess} buildings to process.", idsToProcess.Count);
 
             _ = ScheduleBuildingRepairUpdates(stoppingToken);
             var ticketId = await _ticketing.CreateTicket(null, stoppingToken);
 
             await _sqsRateLimiter.Handle<int>(
-                idsToProcess.ToList(),
+                idsToProcess,
                 id => new RepairBuildingSqsRequest { BuildingPersistentLocalId = id, TicketId = ticketId },
                 async processedId =>
                 {
                     await _repairBuildingRepository.DeleteBuilding(processedId);
-                    if(processedId % 100 == 0)
+                    _logger.LogInformation("Processed building {processedId}", processedId);
+
+                    if (processedId % 100 == 0)
+                    {
                         await WaitIfProducerProjectionBehindAsync(stoppingToken);
+                    }
                 },
                 stoppingToken);
 
@@ -68,7 +73,10 @@ private async Task WaitIfProducerProjectionBehindAsync(CancellationToken cancell
 
                 var lag = headPosition - producerPosition;
                 if (lag <= 100)
+                {
+                    _logger.LogInformation("Resume processing, lag is {Lag} messages.", lag);
                     break;
+                }
 
                 _logger.LogInformation("Producer projection is behind by {Lag} messages, waiting 3 seconds...", lag);
                 await Task.Delay(3000, cancellationToken);