using System; using System.Collections.Generic; using System.Linq; using System.Net.WebSockets; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Hosting; using PolyTrader.Core.Streaming; using PolyTraderSharp.Models; namespace PolyTraderSharp.Services { public class AlchemyWebsocketService : BackgroundService { private const string CtfContractAddress = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045"; private const string TransferSingleTopic = "0xc3d58168c5ae7397731d063d5bbf3d657854427343f4c083240f7aacaa2d0f62"; private const string TransferBatchTopic = "0x4a39dc06d4c0dbc64b70af90fd698a233a518aa5d07e595d983b8c0526c8f7ce"; private readonly TradingState _state; private readonly CopyTradingState _copyState; private readonly ServerSettings _settings; private readonly TraderMonitorService _traderMonitor; private readonly TerminalLogger _logger; private readonly IBlockchainWssClientFactory _wssFactory; public AlchemyWebsocketService( TradingState state, CopyTradingState copyState, ServerSettings settings, TraderMonitorService traderMonitor, TerminalLogger logger, IBlockchainWssClientFactory wssFactory) { _state = state; _copyState = copyState; _settings = settings; _traderMonitor = traderMonitor; _logger = logger; _wssFactory = wssFactory; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { if (!_settings.EnableBlockchainListener || string.IsNullOrEmpty(_settings.PolygonRpcUrl)) { _logger.Info("Blockchain Listener is disabled in settings. Using raw polling."); _state.IsAlchemyHealthy = false; return; } _logger.Info("Alchemy WSS Service starting up..."); while (!stoppingToken.IsCancellationRequested) { if (_state.GlobalTradingPaused || (_state.LiveTradingMode == TradingMode.Inactive && _state.DemoTradingMode == TradingMode.Inactive)) { _state.IsAlchemyHealthy = false; await Task.Delay(5000, stoppingToken); continue; } try { await ConnectAndListenAsync(stoppingToken); } catch (WebSocketException ex) { // Usually indicates a connection drop or 429 _logger.Warning($"Alchemy WSS drop: {ex.Message}. Falling back to API polling for 5 minutes."); _state.IsAlchemyHealthy = false; await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken); } catch (Exception ex) { if (ex.Message.Contains("429") || ex.Message.Contains("Too Many Requests")) { _logger.Error($"Alchemy HTTP 429 Limit reached. Suspending WSS for 5 minutes."); _state.IsAlchemyHealthy = false; await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken); } else { _logger.Error($"Alchemy WSS Error: {ex.Message}. Retrying in 10s..."); _state.IsAlchemyHealthy = false; await Task.Delay(10000, stoppingToken); } } } } /// /// Baut EINE WSS-Session über den Core-Client auf. Consumer-Verantwortung: Filter aus den /// getrackten Master-Tradern bauen, auf Trader-Listen-Änderungen re-subscriben, Health spiegeln. /// private async Task ConnectAndListenAsync(CancellationToken stoppingToken) { var activeTraders = _copyState.Traders.Values.Where(t => t.IsActive).ToList(); var activeStateHash = string.Join(",", activeTraders.OrderBy(t => t.Id).Select(t => t.WalletAddress.ToLowerInvariant())); var subscription = BuildSubscription(activeTraders); if (subscription.Filters.Count == 0) { _logger.Info("Keine aktiven Master-Trader. Socket läuft im Standby..."); _state.IsAlchemyHealthy = false; await Task.Delay(5000, stoppingToken); return; } var client = _wssFactory.Create(); _logger.Info("Connecting to Alchemy WebSocket (Core WSS client)..."); using var loopCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); // Überwacht Trader-Listen-Änderungen (Re-Subscribe) und spiegelt das Health-Flag. var monitorTask = Task.Run(async () => { while (!loopCts.IsCancellationRequested) { try { await Task.Delay(5000, loopCts.Token); } catch (OperationCanceledException) { break; } _state.IsAlchemyHealthy = client.IsHealthy; var currentTraders = _copyState.Traders.Values.Where(t => t.IsActive).ToList(); var currentHash = string.Join(",", currentTraders.OrderBy(t => t.Id).Select(t => t.WalletAddress.ToLowerInvariant())); if (currentHash != activeStateHash) { _logger.Info("🔄 Master-Trader Liste hat sich geändert. Starte Alchemy WSS mit neuen Filtern neu..."); loopCts.Cancel(); break; } } }); try { await client.ConnectAndListenAsync(_settings.PolygonRpcUrl, subscription, OnLog, loopCts.Token); } catch (OperationCanceledException) { // Erwartet beim Re-Subscribe (Trader-Liste geändert) oder beim Shutdown. } finally { _state.IsAlchemyHealthy = false; if (!loopCts.IsCancellationRequested) loopCts.Cancel(); } } private BlockchainWssSubscription BuildSubscription(List activeTraders) { var subscription = new BlockchainWssSubscription(); var paddedAddresses = activeTraders.Select(t => PadAddress(t.WalletAddress)).ToList(); var topic0 = new List { TransferSingleTopic, TransferBatchTopic }; const int batchSize = 3; // Alchemy begrenzt Topic-Arrays auf max. 3-4 Einträge for (int i = 0; i < paddedAddresses.Count; i += batchSize) { var chunk = paddedAddresses.Skip(i).Take(batchSize).ToList(); // Buys: Master-Trader ist Empfänger (Topic 3) subscription.Filters.Add(new LogSubscriptionFilter { Address = CtfContractAddress, Topic0 = topic0, Topic3 = chunk }); // Sells: Master-Trader ist Sender (Topic 2) subscription.Filters.Add(new LogSubscriptionFilter { Address = CtfContractAddress, Topic0 = topic0, Topic2 = chunk }); } return subscription; } private void OnLog(BlockchainLogEvent evt) { if (evt.Topics.Count < 4) return; var fromTopic = evt.Topics[2]?.ToLowerInvariant(); var toTopic = evt.Topics[3]?.ToLowerInvariant(); if (fromTopic == null || toTopic == null) return; var activeTraders = _copyState.Traders.Values.Where(t => t.IsActive).ToList(); string? triggeredAddress = null; foreach (var trader in activeTraders) { var padded = PadAddress(trader.WalletAddress); if (fromTopic == padded || toTopic == padded) { triggeredAddress = trader.WalletAddress; break; } } if (!string.IsNullOrEmpty(triggeredAddress)) { _traderMonitor.TriggerFastBlockchainPoll(evt.TransactionHash, _settings.PolygonRpcUrl, triggeredAddress); } } private string PadAddress(string address) { string stripped = address.Replace("0x", "", StringComparison.OrdinalIgnoreCase).ToLowerInvariant(); return "0x" + stripped.PadLeft(64, '0'); } } }