using Predictalytics.Domain.Entities; using Predictalytics.Domain.Enums; using Predictalytics.Domain.Interfaces; using System.Threading; using Microsoft.EntityFrameworkCore; namespace Predictalytics.Infrastructure.Data.Repositories; public class MarketRepository : IMarketRepository { private readonly AppDbContext _db; private static readonly SemaphoreSlim _syncSemaphore = new(1, 1); public MarketRepository(AppDbContext db) => _db = db; public async Task GetByPlatformIdAsync(PlatformType platform, string platformMarketId, CancellationToken ct = default) => await _db.Markets.Include(m => m.Outcomes) .FirstOrDefaultAsync(m => m.Platform == platform && m.PlatformMarketId == platformMarketId, ct); public async Task GetOutcomeByTokenIdAsync(string tokenId, CancellationToken ct = default) => await _db.MarketOutcomes.Include(o => o.Market) .FirstOrDefaultAsync(o => o.TokenId == tokenId, ct); public async Task> GetOutcomesByTokenIdsAsync(IEnumerable tokenIds, CancellationToken ct = default) => await _db.MarketOutcomes.Include(o => o.Market) .Where(o => tokenIds.Contains(o.TokenId)) .ToListAsync(ct); public async Task AddOrUpdateAsync(Market market, CancellationToken ct = default) { await _syncSemaphore.WaitAsync(ct); try { TruncateMarketStrings(market); var existing = await _db.Markets.Include(m => m.Outcomes) .FirstOrDefaultAsync(m => m.Platform == market.Platform && m.PlatformMarketId == market.PlatformMarketId, ct); if (existing != null) { UpdateMarketFields(existing, market); } else { _db.Markets.Add(market); } await _db.SaveChangesAsync(ct); } finally { _syncSemaphore.Release(); } } public async Task AddOrUpdateRangeAsync(IEnumerable markets, CancellationToken ct = default) { // Deduplicate input by PlatformMarketId to avoid processing the same ID twice in one call var marketList = markets .GroupBy(m => new { m.Platform, m.PlatformMarketId }) .Select(g => g.First()) .ToList(); if (!marketList.Any()) return; await _syncSemaphore.WaitAsync(ct); try { // Process in sub-batches to avoid too large SQL queries const int subBatchSize = 500; for (int i = 0; i < marketList.Count; i += subBatchSize) { var currentBatch = marketList.Skip(i).Take(subBatchSize).ToList(); var platform = currentBatch.First().Platform; var ids = currentBatch.Select(m => m.PlatformMarketId).ToList(); // Fetch all existing markets in this batch at once var existingMarkets = await _db.Markets.Include(m => m.Outcomes) .Where(m => m.Platform == platform && ids.Contains(m.PlatformMarketId)) .ToListAsync(ct); var existingMap = existingMarkets.ToDictionary(m => m.PlatformMarketId); foreach (var market in currentBatch) { TruncateMarketStrings(market); if (existingMap.TryGetValue(market.PlatformMarketId, out var existing)) { UpdateMarketFields(existing, market); } else { _db.Markets.Add(market); } } await _db.SaveChangesAsync(ct); } } finally { _syncSemaphore.Release(); } } private void UpdateMarketFields(Market existing, Market updated) { existing.Question = updated.Question; existing.MarketSlug = updated.MarketSlug; existing.EventSlug = updated.EventSlug; existing.Description = updated.Description; existing.ImageUrl = updated.ImageUrl; existing.Category = updated.Category; existing.Volume = updated.Volume; existing.Liquidity = updated.Liquidity; existing.StartDate = updated.StartDate; existing.EndDate = updated.EndDate; existing.IsResolved = updated.IsResolved; existing.ResolutionOutcome = updated.ResolutionOutcome; existing.CreatedAt = updated.CreatedAt; // Platform creation date existing.LastUpdatedAt = DateTime.UtcNow; // Upsert outcomes foreach (var newOutcome in updated.Outcomes) { var existingOutcome = existing.Outcomes .FirstOrDefault(o => o.OutcomeIndex == newOutcome.OutcomeIndex); if (existingOutcome != null) { existingOutcome.Label = newOutcome.Label; existingOutcome.TokenId = newOutcome.TokenId; existingOutcome.CurrentPrice = newOutcome.CurrentPrice; } else { newOutcome.MarketId = existing.Id; existing.Outcomes.Add(newOutcome); } } } private void TruncateMarketStrings(Market market) { market.Question = StringHelper.Truncate(market.Question, 1024) ?? ""; market.Description = StringHelper.Truncate(market.Description, 4096); market.MarketSlug = StringHelper.Truncate(market.MarketSlug, 512) ?? ""; market.EventSlug = StringHelper.Truncate(market.EventSlug, 512) ?? ""; market.ImageUrl = StringHelper.Truncate(market.ImageUrl, 1024); market.Category = StringHelper.Truncate(market.Category, 128) ?? ""; foreach (var o in market.Outcomes) { o.Label = StringHelper.Truncate(o.Label, 256) ?? ""; } } public async Task> GetActiveAsync(int count = 50, CancellationToken ct = default) => await _db.Markets.Include(m => m.Outcomes) .Where(m => !m.IsResolved) .OrderByDescending(m => m.Volume) .Take(count) .ToListAsync(ct); public async Task GetCountAsync(CancellationToken ct = default) => await _db.Markets.CountAsync(ct); public async Task> GetMarketsDueForTradeUpdateAsync(int cooldownHours, int limit, CancellationToken ct = default) { var cutoff = DateTime.UtcNow.AddHours(-cooldownHours); return await _db.Markets .Where(m => !m.IsResolved && (m.LastTradesUpdatedAt == null || m.LastTradesUpdatedAt < cutoff)) .OrderBy(m => m.LastTradesUpdatedAt ?? DateTime.MinValue) .Take(limit) .ToListAsync(ct); } public async Task UpdateAsync(Market market, CancellationToken ct = default) { TruncateMarketStrings(market); _db.Markets.Update(market); await _db.SaveChangesAsync(ct); } public async Task GetByIdAsync(int id, CancellationToken ct = default) => await _db.Markets.Include(m => m.Outcomes).FirstOrDefaultAsync(m => m.Id == id, ct); public async Task> SearchAsync(string query, int take = 20, CancellationToken ct = default) { if (string.IsNullOrWhiteSpace(query)) return Array.Empty(); return await _db.Markets.Include(m => m.Outcomes) .Where(m => m.Question.Contains(query) || m.PlatformMarketId.Contains(query) || m.Id.ToString() == query) .OrderByDescending(m => m.Volume) .Take(take) .ToListAsync(ct); } public async Task> GetPriceSnapshotsAsync(int marketOutcomeId, CancellationToken ct = default) { return await _db.MarketOutcomePriceSnapshots .Where(ps => ps.MarketOutcomeId == marketOutcomeId) .OrderBy(ps => ps.Timestamp) .ToListAsync(ct); } public async Task SavePriceSnapshotsAsync(int marketOutcomeId, IEnumerable snapshots, CancellationToken ct = default) { var existing = await _db.MarketOutcomePriceSnapshots .Where(ps => ps.MarketOutcomeId == marketOutcomeId) .ToListAsync(ct); _db.MarketOutcomePriceSnapshots.RemoveRange(existing); _db.MarketOutcomePriceSnapshots.AddRange(snapshots); await _db.SaveChangesAsync(ct); } }