Richard/Predictalytics
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Jobs infrastructure, endpoints, UI and worker prioritization

Browse Source
This commit is contained in:
Richard
2026-07-06 19:58:23 +02:00
parent 33a52ed173
commit fb703b6c81
9 changed files with 321 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Predictalytics.Worker/Services/TradeHistoryWorker.cs
+48 -5
View File
@@ -36,20 +36,40 @@ public class TradeHistoryWorker : BackgroundService
{
try
{
IReadOnlyList<Domain.Entities.Trader> tradersToProcess;
IReadOnlyList<Domain.Entities.Trader> tradersToProcess = new List<Domain.Entities.Trader>();
Domain.Entities.BackgroundJob? activeJob = null;
using (var scope = _services.CreateScope())
{
var jobRepo = scope.ServiceProvider.GetRequiredService<IJobRepository>();
activeJob = await jobRepo.GetNextPendingJobAsync(Predictalytics.Domain.Enums.JobType.HistorySync, stoppingToken);
var repo = scope.ServiceProvider.GetRequiredService<ITraderRepository>();
tradersToProcess = await repo.GetTradersDueForTradeUpdateAsync(CooldownHours, TradersPerCycle, stoppingToken);
if (activeJob != null && activeJob.TraderId.HasValue)
{
var t = await repo.GetByIdAsync(activeJob.TraderId.Value, stoppingToken);
if (t != null)
{
tradersToProcess = new[] { t };
activeJob.Status = Predictalytics.Domain.Enums.JobStatus.InProgress;
activeJob.StartedAt = DateTime.UtcNow;
await jobRepo.UpdateAsync(activeJob, stoppingToken);
}
}
else
{
tradersToProcess = await repo.GetTradersDueForTradeUpdateAsync(CooldownHours, TradersPerCycle, stoppingToken);
}
}
if (tradersToProcess.Count == 0)
{
_logger.LogDebug("📜 No traders due for sync. Sleeping.");
_logger.LogDebug("🔄 No traders due for sync. Sleeping.");
}
else
{
_logger.LogInformation("📜 Processing {Count} traders for trade sync (Initial/Update)", tradersToProcess.Count);
_logger.LogInformation("🔄 Processing {Count} traders for trade sync (Initial/Update)", tradersToProcess.Count);
}
var outcomeCache = new System.Collections.Concurrent.ConcurrentDictionary<string, Domain.Entities.MarketOutcome>();
@@ -166,9 +186,32 @@ public class TradeHistoryWorker : BackgroundService
trader.LastTradesUpdatedAt = DateTime.UtcNow;
trader.LastPolledAt = DateTime.UtcNow;
await traderRepo.UpdateAsync(trader, ct);
if (activeJob != null && activeJob.TraderId == trader.Id)
{
using var jobScope = _services.CreateScope();
var updateJobRepo = jobScope.ServiceProvider.GetRequiredService<IJobRepository>();
activeJob.Status = Predictalytics.Domain.Enums.JobStatus.Completed;
activeJob.CompletedAt = DateTime.UtcNow;
await updateJobRepo.UpdateAsync(activeJob, ct);
}
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { /* App shutting down */ }
catch (Exception ex) { _logger.LogError(ex, "Error syncing trader {TraderId}", t.Id); }
catch (Exception ex)
{
_logger.LogError(ex, "Error syncing trader {TraderId}", t.Id);
if (activeJob != null && activeJob.TraderId == t.Id)
{
try {
using var jobScope = _services.CreateScope();
var updateJobRepo = jobScope.ServiceProvider.GetRequiredService<IJobRepository>();
activeJob.Status = Predictalytics.Domain.Enums.JobStatus.Failed;
activeJob.CompletedAt = DateTime.UtcNow;
activeJob.ErrorMessage = ex.Message;
await updateJobRepo.UpdateAsync(activeJob, CancellationToken.None);
} catch { /* Ignore secondary errors */ }
}
}
});
}
catch (OperationCanceledException) { break; }