using System.Collections.Concurrent; using Marathon.Application.Abstractions; using Marathon.Application.Configuration; using Marathon.Domain.Entities; using Marathon.Domain.Enums; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; namespace Marathon.Application.UseCases; /// /// Fetches the current pre-match event listing, persists new events (skipping /// duplicates by ), and captures an /// initial pre-match odds snapshot for every returned event. /// public sealed class PullUpcomingEventsUseCase { private readonly IOddsScraper _scraper; private readonly IEventRepository _eventRepo; private readonly ISnapshotRepository _snapshotRepo; private readonly IOptionsMonitor _throttle; private readonly ILogger _logger; public PullUpcomingEventsUseCase( IOddsScraper scraper, IEventRepository eventRepo, ISnapshotRepository snapshotRepo, IOptionsMonitor throttle, ILogger logger) { _scraper = scraper ?? throw new ArgumentNullException(nameof(scraper)); _eventRepo = eventRepo ?? throw new ArgumentNullException(nameof(eventRepo)); _snapshotRepo = snapshotRepo ?? throw new ArgumentNullException(nameof(snapshotRepo)); _throttle = throttle ?? throw new ArgumentNullException(nameof(throttle)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); } /// /// Executes one polling cycle: scrape → persist new events → capture snapshots. /// /// Cancellation token. /// /// A tuple of (EventsProcessed, NewEvents, SnapshotsCaptured). /// EventsProcessed is the total number returned by the scraper. /// NewEvents is how many were not already in the DB. /// SnapshotsCaptured is how many snapshots were successfully saved. /// public async Task<(int EventsProcessed, int NewEvents, int SnapshotsCaptured)> ExecuteAsync( CancellationToken ct) { _logger.LogInformation("PullUpcomingEventsUseCase: cycle started"); var events = await _scraper.ScrapeUpcomingAsync(sportFilter: null, ct); int eventsProcessed = events.Count; _logger.LogInformation( "PullUpcomingEventsUseCase: scraper returned {Count} events", eventsProcessed); // Phase 1 — parallel HTTP fan-out. Each event's odds snapshot is scraped // concurrently up to MaxConcurrentRequests; the scraper's rate limiter // smooths spikes underneath. We do NOT touch the DbContext here — EF Core // is single-threaded. var scraped = new ConcurrentBag<(Event Event, OddsSnapshot Snapshot)>(); var maxParallelism = Math.Max(1, _throttle.CurrentValue.MaxConcurrentRequests); var parallelOptions = new ParallelOptions { MaxDegreeOfParallelism = maxParallelism, CancellationToken = ct, }; await Parallel.ForEachAsync(events, parallelOptions, async (ev, taskCt) => { try { var snapshot = await _scraper.ScrapeEventOddsAsync(ev, OddsSource.PreMatch, taskCt); scraped.Add((ev, snapshot)); } catch (OperationCanceledException) { throw; } catch (Exception ex) { _logger.LogWarning(ex, "PullUpcomingEventsUseCase: failed to capture snapshot for event {EventId} — skipping", ev.Id.Value); } }); // Phase 2 — sequential persistence. Upsert event row, then save the // captured snapshot. Per-event try/catch keeps a single failure from // aborting the whole cycle. int newEvents = 0; int snapshotsCaptured = 0; foreach (var (ev, snapshot) in scraped) { ct.ThrowIfCancellationRequested(); try { var existing = await _eventRepo.GetAsync(ev.Id, ct); if (existing is null) { await _eventRepo.AddAsync(ev, ct); await _eventRepo.SaveChangesAsync(ct); newEvents++; } } catch (OperationCanceledException) { throw; } catch (Exception ex) { _logger.LogWarning(ex, "PullUpcomingEventsUseCase: failed to persist event {EventId} — skipping", ev.Id.Value); } try { await _snapshotRepo.AddAsync(snapshot, ct); await _snapshotRepo.SaveChangesAsync(ct); snapshotsCaptured++; } catch (OperationCanceledException) { throw; } catch (Exception ex) { _logger.LogWarning(ex, "PullUpcomingEventsUseCase: failed to persist snapshot for event {EventId} — skipping", ev.Id.Value); } } _logger.LogInformation( "PullUpcomingEventsUseCase: cycle done — processed={Processed}, new={New}, snapshots={Snapshots}", eventsProcessed, newEvents, snapshotsCaptured); return (eventsProcessed, newEvents, snapshotsCaptured); } }