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).Include(m => m.Event) .FirstOrDefaultAsync(m => m.Platform == platform && m.ConditionId == 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.ConditionId == market.ConditionId, ct); if (existing != null) { UpdateMarketFields(existing, market); } else { if (market.Event == null && market.EventId == 0) { // Fallback to avoid foreign key exceptions if event is entirely missing market.Event = new Event { Platform = market.Platform, PlatformEventId = market.PlatformMarketId, Slug = "unknown", Title = "Unknown" }; } _db.Markets.Add(market); } await _db.SaveChangesAsync(ct); } finally { _syncSemaphore.Release(); } } public async Task AddOrUpdateRangeAsync(IEnumerable markets, CancellationToken ct = default) { // Deduplicate input by ConditionId to avoid processing the same ID twice in one call var marketList = markets .GroupBy(m => new { m.Platform, m.ConditionId }) .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.ConditionId).ToList(); var existingMarkets = await _db.Markets.Include(m => m.Outcomes) .Where(m => m.Platform == platform && ids.Contains(m.ConditionId)) .ToListAsync(ct); var existingMap = existingMarkets.ToDictionary(m => m.ConditionId); foreach (var market in currentBatch) { TruncateMarketStrings(market); if (existingMap.TryGetValue(market.ConditionId, out var existing)) { UpdateMarketFields(existing, market); } else { if (market.Event == null && market.EventId == 0) { market.Event = new Event { Platform = market.Platform, PlatformEventId = market.PlatformMarketId, Slug = "unknown", Title = "Unknown" }; } _db.Markets.Add(market); } } await _db.SaveChangesAsync(ct); } } finally { _syncSemaphore.Release(); } } public async Task AddOrUpdateEventsAsync(IEnumerable events, CancellationToken ct = default) { var eventList = events.GroupBy(e => new { e.Platform, e.PlatformEventId }).Select(g => g.First()).ToList(); if (!eventList.Any()) return; await _syncSemaphore.WaitAsync(ct); try { const int subBatchSize = 100; for (int i = 0; i < eventList.Count; i += subBatchSize) { var currentBatch = eventList.Skip(i).Take(subBatchSize).ToList(); var platform = currentBatch.First().Platform; var eventIds = currentBatch.Select(e => e.PlatformEventId).ToList(); var existingEvents = await _db.Events .Include(e => e.Markets).ThenInclude(m => m.Outcomes) .Where(e => e.Platform == platform && eventIds.Contains(e.PlatformEventId)) .ToListAsync(ct); var existingEventsMap = existingEvents.ToDictionary(e => e.PlatformEventId); foreach (var ev in currentBatch) { if (ev.Slug != null && ev.Slug.Length > 512) ev.Slug = ev.Slug[..512]; if (ev.Title != null && ev.Title.Length > 1024) ev.Title = ev.Title[..1024]; if (existingEventsMap.TryGetValue(ev.PlatformEventId, out var existing)) { existing.Slug = ev.Slug!; existing.Title = ev.Title!; existing.Description = ev.Description; existing.ImageUrl = ev.ImageUrl; existing.Tags = ev.Tags; existing.StartDate = ev.StartDate; existing.EndDate = ev.EndDate; existing.IsActive = ev.IsActive; existing.IsClosed = ev.IsClosed; existing.LastUpdatedAt = DateTime.UtcNow; // Upsert markets inside event foreach (var market in ev.Markets) { TruncateMarketStrings(market); var existingMarket = existing.Markets.FirstOrDefault(m => m.ConditionId == market.ConditionId); if (existingMarket != null) { UpdateMarketFields(existingMarket, market); } else { market.EventId = existing.Id; market.Event = null!; // Prevent EF tracking issue existing.Markets.Add(market); } } } else { foreach (var m in ev.Markets) TruncateMarketStrings(m); _db.Events.Add(ev); } } await _db.SaveChangesAsync(ct); } } finally { _syncSemaphore.Release(); } } private void UpdateMarketFields(Market existing, Market updated) { existing.Question = updated.Question; existing.MarketSlug = updated.MarketSlug; existing.PlatformMarketId = updated.PlatformMarketId; existing.QuestionId = updated.QuestionId; existing.Description = updated.Description; // The /markets endpoint (on-demand path) carries no event tags, so its // classification is often just "Other". Never let an uninformative update // overwrite a category previously derived from the tag-bearing /events sync. if (updated.Category != MarketCategory.Other || existing.Category == MarketCategory.Other) { existing.Category = updated.Category; existing.Subcategory = updated.Subcategory; } existing.Volume = updated.Volume; existing.Volume24h = updated.Volume24h; existing.Liquidity = updated.Liquidity; existing.StartDate = updated.StartDate; existing.EndDate = updated.EndDate; existing.IsResolved = updated.IsResolved; existing.ResolutionOutcome = updated.ResolutionOutcome; existing.CreatedAt = updated.CreatedAt; 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.ImageUrl = StringHelper.Truncate(market.ImageUrl, 1024); market.Subcategory = StringHelper.Truncate(market.Subcategory, 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.ConditionId.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); } }