Predictalytics/src/Predictalytics.Worker/Services/MarketHistoryWorker.cs

using Predictalytics.Application.Interfaces;
using Predictalytics.Domain.Enums;
using Predictalytics.Domain.Interfaces;
using Predictalytics.Infrastructure.Logging;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace Predictalytics.Worker.Services;

/// <summary>
/// Background service that periodically fetches recent trades for active markets
/// to discover new unknown traders. Updates markets starting with the oldest refreshed.
/// </summary>
public class MarketHistoryWorker : BackgroundService
{
    private readonly IServiceProvider _services;
    private readonly ILogger<MarketHistoryWorker> _logger;

    private const int CooldownHours = 6;
    private const int MarketsPerCycle = 10;
    private const int TradesPerFetch = 100;

    public MarketHistoryWorker(IServiceProvider services, ILogger<MarketHistoryWorker> logger)
    { _services = services; _logger = logger; }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("🕵️ MarketHistoryWorker started (cooldown: {Hours}h)", CooldownHours);
        await Task.Delay(25000, stoppingToken);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                using var scope = _services.CreateScope();
                var marketRepo = scope.ServiceProvider.GetRequiredService<IMarketRepository>();
                var traderRepo = scope.ServiceProvider.GetRequiredService<ITraderRepository>();
                var providers = scope.ServiceProvider.GetRequiredService<IEnumerable<IPlatformProvider>>();
                var rateLimiter = scope.ServiceProvider.GetRequiredService<IRateLimiter>();

                var markets = await marketRepo.GetMarketsDueForTradeUpdateAsync(CooldownHours, MarketsPerCycle, stoppingToken);

                if (markets.Count == 0)
                {
                    _logger.LogDebug("🕵️ No markets due for trade history update. Sleeping.");
                }
                else
                {
                    _logger.LogInformation("🕵️ Processing {Count} markets for trade history sync", markets.Count);
                }

                int totalDiscovered = 0;

                foreach (var market in markets)
                {
                    if (stoppingToken.IsCancellationRequested) break;

                    var provider = providers.FirstOrDefault(p => p.Platform == market.Platform && p.IsImplemented);
                    if (provider == null) continue;

                    try
                    {
                        using var platformCtx = PlatformLogContext.Push(provider.PlatformName);
                        await rateLimiter.WaitAsync(market.Platform, stoppingToken);

                        var trades = await provider.GetMarketTradesAsync(market.PlatformMarketId, TradesPerFetch, stoppingToken);

                        // Extract unique wallet addresses from TransientWallet [NotMapped].
                        // Providers set this during in-memory mapping; it is NOT stored in the DB.
                        var wallets = trades
                            .Select(t => t.TransientWallet)
                            .Where(w => !string.IsNullOrWhiteSpace(w))
                            .Select(w => w!)
                            .Distinct()
                            .ToList();

                        foreach (var wallet in wallets)
                        {
                            var existing = await traderRepo.GetByPlatformIdAsync(
                                market.Platform, wallet, stoppingToken);

                            if (existing == null)
                            {
                                var trader = new Domain.Entities.Trader
                                {
                                    Platform = market.Platform,
                                    PlatformUserId = wallet,
                                    DisplayName = wallet.Length > 8 ? wallet[..8] + "..." : wallet, // Will be updated later
                                    IsAutoDiscovered = true,
                                    CreatedAt = DateTime.UtcNow
                                };

                                await traderRepo.AddAsync(trader, stoppingToken);
                                totalDiscovered++;

                                _logger.LogInformation("[{Platform}] Discovered trader ({Wallet}) from market trades: {Market}",
                                    provider.PlatformName, wallet[..10] + "...",
                                    market.Question.Length > 60 ? market.Question[..60] + "..." : market.Question);
                            }
                        }

                        // Update timestamps
                        market.LastTradesUpdatedAt = DateTime.UtcNow;
                        await marketRepo.UpdateAsync(market, stoppingToken);
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "Error syncing trade history for market {Market} on {Platform}",
                            market.PlatformMarketId, market.Platform);
                    }
                }

                if (totalDiscovered > 0)
                {
                    _logger.LogInformation("✅ Market history cycle complete. {Count} new traders discovered.", totalDiscovered);
                }
            }
            catch (OperationCanceledException) { break; }
            catch (Exception ex) { _logger.LogError(ex, "MarketHistoryWorker error"); }

            // Run every 2 minutes
            await Task.Delay(TimeSpan.FromMinutes(2), stoppingToken);
        }

        _logger.LogInformation("🕵️ MarketHistoryWorker stopped");
    }
}