<?php
/**
 * WebSocket Server para WebTrader
 * Servidor WebSocket con integración de Finage API para datos en tiempo real
 */

require_once __DIR__ . '/../vendor/autoload.php';
require_once __DIR__ . '/../../../database/config.php';
require_once __DIR__ . '/../finage_api.php';

use React\EventLoop\Loop;
use React\Socket\SocketServer;
use React\Stream\WritableResourceStream;
// RFC6455 WebSocket handshake y framing (sin servidor Ratchet completo)
use GuzzleHttp\Psr7\Message as PsrMessage;
use GuzzleHttp\Psr7\Response as PsrResponse;
use GuzzleHttp\Psr7\HttpFactory;
use Ratchet\RFC6455\Handshake\RequestVerifier;
use Ratchet\RFC6455\Handshake\ServerNegotiator;
use Ratchet\RFC6455\Handshake\PermessageDeflateOptions;
use Ratchet\RFC6455\Messaging\CloseFrameChecker;
use Ratchet\RFC6455\Messaging\MessageBuffer;
use Ratchet\RFC6455\Messaging\MessageInterface as WsMessageInterface;
use Ratchet\RFC6455\Messaging\FrameInterface as WsFrameInterface;
use Ratchet\RFC6455\Messaging\Frame as WsFrame;

class WebTraderWebSocketServer {
    private $loop;
    private $clients = [];
    private $clientData = [];
    private $subscriptions = [];
    private $clientParsers = [];
    private $db;
    private $finageAPI;
    // Estado de conexión WS de Finage y símbolos FOREX suscritos
    private $finageWsConn = null; // instancia de Ratchet\Client\WebSocket
    private $finageSubscribedForex = [];
    // Agregadores OHLC por símbolo y timeframe (minutos)
    private $candleAggregators = [];
    private $candleTimeframes = [1,5,15,30,60,240,1440];
    
    public function __construct() {
        $this->loop = Loop::get();
        
        // Inicializar Finage API
        $this->finageAPI = getFinageAPI();
        
        // Conectar a la base de datos
        try {
            $this->db = new PDO("mysql:host=" . DB_HOST . ";dbname=" . DB_NAME, DB_USER, DB_PASS);
            $this->db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
            echo "✅ Conexión a base de datos establecida\n";
        } catch (PDOException $e) {
            echo "❌ Error conectando a la base de datos: " . $e->getMessage() . "\n";
        }
        
        // Test Finage API connection
        $testResult = $this->finageAPI->testConnection();
        if ($testResult['success']) {
            echo "✅ Conexión a Finage API establecida\n";
        } else {
            echo "⚠️  Finage API no disponible, sin feed en tiempo real\n";
        }
        
        echo "🚀 WebSocket Server iniciado\n";
    }
    
    public function start($port = 8082) {
        $socket = new SocketServer("0.0.0.0:$port", [], $this->loop);

        // Configurar handshake RFC6455 y parser de frames
        $httpFactory = new HttpFactory();
        $negotiator = new ServerNegotiator(new RequestVerifier(), $httpFactory, PermessageDeflateOptions::permessageDeflateSupported());
        $closeFrameChecker = new CloseFrameChecker();
        $uException = new \UnderflowException();

        $socket->on('connection', function (\React\Socket\ConnectionInterface $connection) use ($negotiator, $closeFrameChecker, $uException) {
            $clientId = uniqid();
            $this->clients[$clientId] = $connection;
            $this->clientData[$clientId] = [
                'authenticated' => false,
                'client_id' => null,
                'subscriptions' => []
            ];

            echo "🔗 Nueva conexión: $clientId\n";

            $headerComplete = false;
            $buffer = '';
            $parser = null;

            $connection->on('data', function ($data) use ($connection, &$parser, &$headerComplete, &$buffer, $negotiator, $closeFrameChecker, $uException, $clientId) {
                if ($headerComplete) {
                    // Frames WebSocket después del handshake
                    $parser->onData($data);
                    return;
                }

                // Acumular cabeceras HTTP del handshake
                $buffer .= $data;
                $parts = explode("\r\n\r\n", $buffer);
                if (count($parts) < 2) {
                    return;
                }
                $headerComplete = true;

                $psrRequest = PsrMessage::parseRequest($parts[0] . "\r\n\r\n");
                $negotiatorResponse = $negotiator->handshake($psrRequest)->withAddedHeader('Content-Length', '0');

                // Responder al handshake
                $connection->write(PsrMessage::toString($negotiatorResponse));

                if ($negotiatorResponse->getStatusCode() !== 101) {
                    // Handshake fallido
                    $connection->end();
                    return;
                }

                // Configurar parser de mensajes WebSocket
                $deflateOptions = PermessageDeflateOptions::fromRequestOrResponse($psrRequest)[0];
                $parser = new MessageBuffer(
                    $closeFrameChecker,
                    // Mensaje de texto/binario recibido
                    function (WsMessageInterface $message, MessageBuffer $messageBuffer) use ($clientId) {
                        $payload = $message->getPayload();
                        $this->handleMessage($clientId, $payload);
                    },
                    // Control frames
                    function (WsFrameInterface $frame) use ($connection, &$parser) {
                        switch ($frame->getOpCode()) {
                            case WsFrame::OP_CLOSE:
                                $connection->end($frame->getContents());
                                break;
                            case WsFrame::OP_PING:
                                $connection->write($parser->newFrame($frame->getPayload(), true, WsFrame::OP_PONG)->getContents());
                                break;
                        }
                    },
                    true,
                    function (): \Exception { return $uException; },
                    null,
                    null,
                    [$connection, 'write'],
                    $deflateOptions
                );

                // Guardar parser para enviar mensajes posteriormente
                $this->clientParsers[$clientId] = $parser;

                // Enviar bienvenida como mensaje WebSocket
                $this->sendMessageToClientId($clientId, [
                    'type' => 'welcome',
                    'message' => 'Conectado al WebTrader WebSocket Server',
                    'connection_id' => $clientId
                ]);

                // Si el cliente envió datos tras el handshake, procesarlos
                array_shift($parts);
                if (!empty($parts)) {
                    $parser->onData(implode("\r\n\r\n", $parts));
                }
            });

            $connection->on('close', function () use ($clientId) {
                $this->handleDisconnection($clientId);
            });

            $connection->on('error', function ($error) use ($clientId) {
                echo "❌ Error en conexión $clientId: " . $error->getMessage() . "\n";
                $this->handleDisconnection($clientId);
            });
        });
        
        // Iniciar actualizaciones automáticas de precios
        $this->startPriceUpdates();
        // Iniciar dispatcher de eventos desde BD
        $this->startEventDispatcher();
        
        echo "🌐 WebSocket Server corriendo en ws://localhost:$port\n";
        echo "📊 Actualizaciones de precios cada 1 segundo\n";
        echo "🔄 Presiona Ctrl+C para detener\n";
        
        $this->loop->run();
    }
    
    private function handleMessage($clientId, $data) {
        try {
            // El payload ya es el texto del mensaje WebSocket
            $message = json_decode($data, true);
            
            if (!$message || !isset($message['type'])) {
                $this->sendError($clientId, 'Formato de mensaje inválido');
                return;
            }
            
            switch ($message['type']) {
                case 'authenticate':
                    $this->handleAuthentication($clientId, $message);
                    break;
                    
                case 'subscribe':
                    $this->handleSubscription($clientId, $message);
                    break;
                    
                case 'unsubscribe':
                    $this->handleUnsubscription($clientId, $message);
                    break;
                    
                case 'ping':
                    $this->handlePing($clientId, $message);
                    break;
                    
                default:
                    $this->sendError($clientId, 'Tipo de mensaje no reconocido');
            }
            
        } catch (Exception $e) {
            echo "❌ Error procesando mensaje: " . $e->getMessage() . "\n";
            $this->sendError($clientId, 'Error interno del servidor');
        }
    }
    
    private function handleAuthentication($clientId, $data) {
        if (!isset($data['client_id'])) {
            $this->sendError($clientId, 'client_id requerido para autenticación');
            return;
        }
        
        $clientDbId = $data['client_id'];
        
        // Verificar que el cliente existe en la base de datos
        try {
            $stmt = $this->db->prepare("SELECT id, first_name, last_name, email FROM clients WHERE id = ? AND status = 'active'");
            $stmt->execute([$clientDbId]);
            $client = $stmt->fetch(PDO::FETCH_ASSOC);
            
            if (!$client) {
                $this->sendError($clientId, 'Cliente no encontrado o inactivo');
                return;
            }
            
            // Marcar como autenticado
            $this->clientData[$clientId]['authenticated'] = true;
            $this->clientData[$clientId]['client_id'] = $clientDbId;
            
            $this->sendMessage($this->clients[$clientId], [
                'type' => 'authenticated',
                'success' => true,
                'client_data' => $client
            ]);
            
            echo "✅ Cliente autenticado: $clientDbId (conexión $clientId)\n";
            
        } catch (PDOException $e) {
            echo "❌ Error en autenticación: " . $e->getMessage() . "\n";
            $this->sendError($clientId, 'Error interno de autenticación');
        }
    }
    
    private function handleSubscription($clientId, $data) {
        // Permitir suscripciones limitadas en modo invitado (sin autenticación)
        $guestAllowedChannels = ['prices', 'candle_update'];
        $isAuthenticated = $this->clientData[$clientId]['authenticated'];
        $channel = isset($data['channel']) ? $data['channel'] : null;
        $symbol = isset($data['symbol']) ? $data['symbol'] : null;
        $timeframe = isset($data['timeframe']) ? intval($data['timeframe']) : null;

        if (!$isAuthenticated) {
            if (!$channel || !in_array($channel, $guestAllowedChannels)) {
                $this->sendError($clientId, 'Debe autenticarse primero');
                return;
            }
            // Validar símbolo permitido para invitados
            $config = include(__DIR__ . '/../../../config/finage_config.php');
            $allowedGuestSymbols = array_merge(
                $config['default_symbols'] ?? [],
                $config['crypto_symbols'] ?? []
            );
            if ($symbol && !in_array($symbol, $allowedGuestSymbols)) {
                $this->sendError($clientId, 'Símbolo no permitido en modo invitado');
                return;
            }
            // Validar timeframe permitido
            if ($channel === 'candle_update') {
                if (!$symbol || !$timeframe) {
                    $this->sendError($clientId, 'Se requieren símbolo y timeframe para velas');
                    return;
                }
                if (!in_array($timeframe, $this->candleTimeframes)) {
                    $this->sendError($clientId, 'Timeframe no permitido');
                    return;
                }
            }
        }
        if (!isset($data['channel'])) {
            $this->sendError($clientId, 'Canal requerido para suscripción');
            return;
        }
        
        // $channel, $symbol, $timeframe ya definidos arriba

        // Crear clave de canal con símbolo si es necesario
        // Compatibilidad: mapear alias de canales
        if ($channel === 'prices' || $channel === 'price_update') {
            if (!$symbol) {
                $this->sendError($clientId, 'Se requiere símbolo para suscripción de precios');
                return;
            }
            $channelKey = "prices:{$symbol}";
            $channel = 'prices';
        } else if ($channel === 'account' || $channel === 'account_update') {
            $channelKey = 'account_update';
            $channel = 'account_update';
            // Permitir que el cliente indique la cuenta activa
            if (!empty($symbol)) {
                $this->clientData[$clientId]['active_account_number'] = $symbol;
            }
        } else if ($channel === 'candle_update') {
            if (!$symbol || !$timeframe) {
                $this->sendError($clientId, 'Se requieren símbolo y timeframe para velas');
                return;
            }
            $channelKey = "candle_update:{$symbol}:{$timeframe}";
        } else {
            $channelKey = $symbol ? "{$channel}:{$symbol}" : $channel;
        }
        
        // Agregar a suscripciones
        if (!isset($this->subscriptions[$channelKey])) {
            $this->subscriptions[$channelKey] = [];
        }
        
        $this->subscriptions[$channelKey][$clientId] = $this->clients[$clientId];
        $this->clientData[$clientId]['subscriptions'][] = $channelKey;
        
        $this->sendMessage($this->clients[$clientId], [
            'type' => 'subscribed',
            'channel' => $channel,
            'symbol' => $symbol,
            'success' => true
        ]);
        
        // Enviar datos iniciales según el canal
        $this->sendInitialData($clientId, $channel, $symbol, $timeframe);
        // Suscripción LIVE en Finage WS si procede
        if ($channel === 'prices' && $symbol) {
            $this->ensureFinageSubscription($symbol);
        }

        echo "📡 Cliente $clientId suscrito a: $channelKey\n";
    }
    
    private function handleUnsubscription($clientId, $data) {
        if (!isset($data['channel'])) {
            $this->sendError($clientId, 'Canal requerido para desuscripción');
            return;
        }
        
        $channel = $data['channel'];
        $symbol = isset($data['symbol']) ? $data['symbol'] : null;
        $channelKey = $symbol ? "{$channel}:{$symbol}" : $channel;
        
        // Remover de suscripciones
        if (isset($this->subscriptions[$channelKey][$clientId])) {
            unset($this->subscriptions[$channelKey][$clientId]);
            
            if (empty($this->subscriptions[$channelKey])) {
                unset($this->subscriptions[$channelKey]);
            }
        }
        
        // Remover de datos del cliente
        $clientSubscriptions = &$this->clientData[$clientId]['subscriptions'];
        $key = array_search($channelKey, $clientSubscriptions);
        if ($key !== false) {
            unset($clientSubscriptions[$key]);
        }
        
        $this->sendMessage($this->clients[$clientId], [
            'type' => 'unsubscribed',
            'channel' => $channel,
            'symbol' => $symbol,
            'success' => true
        ]);
        
        echo "📡 Cliente $clientId desuscrito de: $channelKey\n";
    }
    
    private function handlePing($clientId, $data) {
        $this->sendMessage($this->clients[$clientId], [
            'type' => 'pong',
            'timestamp' => time()
        ]);
    }
    
    private function handleDisconnection($clientId) {
        echo "🔌 Conexión cerrada: $clientId\n";
        
        // Limpiar suscripciones
        if (isset($this->clientData[$clientId])) {
            $clientData = $this->clientData[$clientId];
            
            foreach ($clientData['subscriptions'] as $channel) {
                if (isset($this->subscriptions[$channel])) {
                    unset($this->subscriptions[$channel][$clientId]);
                    if (empty($this->subscriptions[$channel])) {
                        unset($this->subscriptions[$channel]);
                    }
                    
                }
            }
            
            unset($this->clientData[$clientId]);
        }
        
        unset($this->clients[$clientId]);
        if (isset($this->clientParsers[$clientId])) {
            unset($this->clientParsers[$clientId]);
        }
    }
    
    private function sendInitialData($clientId, $channel, $symbol = null, $timeframe = null) {
        $clientDbId = $this->clientData[$clientId]['client_id'];
        
        try {
            switch ($channel) {
                case 'prices':
                    if ($symbol) {
                        $rt = $this->getRealTimePrice($symbol);
                        $this->sendMessageToClientId($clientId, [
                            'type' => 'price_update',
                            'symbol' => $symbol,
                            'bid' => $rt['bid'],
                            'ask' => $rt['ask'],
                            'timestamp' => round(microtime(true) * 1000)
                        ]);
                    }
                    break;
                
                case 'account_update':
                    $activeAcc = isset($this->clientData[$clientId]['active_account_number']) ? $this->clientData[$clientId]['active_account_number'] : null;
                    $accountData = null;
                    if ($activeAcc) {
                        $accountData = $this->getAccountDataExtendedByAccountNumber($clientDbId, $activeAcc);
                        if (!$accountData) {
                            // Fallback a datos del portal del cliente
                            $accountData = $this->getClientPortalAccountData($clientDbId, $activeAcc);
                        }
                    }
                    if (!$accountData) {
                        $accountData = $this->getAccountDataExtended($clientDbId);
                    }
                    if ($accountData) {
                        $this->sendMessageToClientId($clientId, [
                            'type' => 'account_update',
                            'data' => $accountData
                        ]);
                    }
                    break;
                
                case 'candle_update':
                    if ($symbol && $timeframe) {
                        $tfCode = $this->mapMinutesToTfCode($timeframe);
                        $candles = $this->fetchInitialCandles($symbol, $tfCode);
                        $this->sendMessageToClientId($clientId, [
                            'type' => 'candle_snapshot',
                            'symbol' => $symbol,
                            'timeframe' => $timeframe,
                            'candles' => $candles
                        ]);
                    }
                    break;
            }
        } catch (Exception $e) {
            echo "❌ Error enviando datos iniciales: " . $e->getMessage() . "\n";
        }
    }

    // Versión extendida que incluye número de cuenta y metadatos
    private function getAccountDataExtended($clientId) {
        try {
            $stmt = $this->db->prepare("SELECT account_number, balance, equity, margin, free_margin, margin_level, currency, leverage FROM trading_accounts WHERE client_id = ? AND status = 'active' LIMIT 1");
            $stmt->execute([$clientId]);
            return $stmt->fetch(PDO::FETCH_ASSOC);
        } catch (PDOException $e) {
            echo "❌ Error obteniendo datos de cuenta extendidos: " . $e->getMessage() . "\n";
            return null;
        }
    }

    // Obtener datos de cuenta por número de cuenta (preferencia)
    private function getAccountDataExtendedByAccountNumber($clientId, $accountNumber) {
        try {
            $stmt = $this->db->prepare("SELECT account_number, balance, equity, margin, free_margin, margin_level, currency, leverage FROM trading_accounts WHERE client_id = ? AND account_number = ? AND status = 'active' LIMIT 1");
            $stmt->execute([$clientId, $accountNumber]);
            $row = $stmt->fetch(PDO::FETCH_ASSOC);
            return $row ?: null;
        } catch (PDOException $e) {
            echo "❌ Error obteniendo datos de cuenta por número: " . $e->getMessage() . "\n";
            return null;
        }
    }

    // Fallback: datos desde client_accounts si no existe en trading_accounts
    private function getClientPortalAccountData($clientId, $accountNumber) {
        try {
            $stmt = $this->db->prepare("SELECT account_number, balance, currency FROM client_accounts WHERE client_id = ? AND account_number = ? AND status = 'active' LIMIT 1");
            $stmt->execute([$clientId, $accountNumber]);
            $row = $stmt->fetch(PDO::FETCH_ASSOC);
            if (!$row) return null;
            // Mapear a estructura esperada por WebTrader
            return [
                'account_number' => $row['account_number'],
                'balance' => (float)$row['balance'],
                'equity' => (float)$row['balance'],
                'margin' => 0.0,
                'free_margin' => (float)$row['balance'],
                'margin_level' => 0.0,
                'currency' => $row['currency'] ?: 'USD',
                'leverage' => 100
            ];
        } catch (PDOException $e) {
            echo "❌ Error obteniendo datos de client portal: " . $e->getMessage() . "\n";
            return null;
        }
    }
    
    private function getAccountData($clientId) {
        try {
            $stmt = $this->db->prepare("
                SELECT balance, equity, margin, free_margin, margin_level 
                FROM trading_accounts 
                WHERE client_id = ? AND status = 'active'
            ");
            $stmt->execute([$clientId]);
            return $stmt->fetch(PDO::FETCH_ASSOC);
        } catch (PDOException $e) {
            echo "❌ Error obteniendo datos de cuenta: " . $e->getMessage() . "\n";
            return null;
        }
    }
    
    private function getRealTimePrice($symbol) {
        // Try to get real price from Finage API
        $realPrice = $this->finageAPI->getRealTimePrice($symbol);
        
        if ($realPrice) {
            return [
                'bid' => number_format($realPrice['bid'], 5),
                'ask' => number_format($realPrice['ask'], 5),
                'change' => $realPrice['change'] ?? 0,
                'change_percent' => $realPrice['change_percent'] ?? 0,
                'source' => 'finage'
            ];
        }
        
        // Sin fallback simulado: devolver null si no hay datos reales
        return null;
    }
    
    private function sendMessage($connection, $data) {
        $clientId = $this->findClientIdByConnection($connection);
        if ($clientId && isset($this->clientParsers[$clientId])) {
            $this->clientParsers[$clientId]->sendMessage(json_encode($data), true, false);
            return;
        }
        // Fallback
        $connection->write(json_encode($data) . "\n");
    }

    private function sendMessageToClientId($clientId, $data) {
        if (isset($this->clientParsers[$clientId])) {
            $this->clientParsers[$clientId]->sendMessage(json_encode($data), true, false);
            return;
        }
        if (isset($this->clients[$clientId])) {
            $this->clients[$clientId]->write(json_encode($data) . "\n");
        }
    }

    private function findClientIdByConnection($connection) {
        foreach ($this->clients as $cid => $conn) {
            if ($conn === $connection) {
                return $cid;
            }
        }
        return null;
    }
    
    private function sendError($clientId, $message) {
        if (isset($this->clients[$clientId])) {
            $this->sendMessage($this->clients[$clientId], [
                'type' => 'error',
                'message' => $message
            ]);
        }
    }
    
    private function broadcast($channel, $data) {
        if (!isset($this->subscriptions[$channel])) {
            return;
        }
        
        foreach ($this->subscriptions[$channel] as $connection) {
            $this->sendMessage($connection, $data);
        }
    }

    /**
     * Dispatcher de eventos en tiempo real desde la base de datos
     * Lee la cola webtrader_events y emite actualizaciones a los canales suscritos
     */
    private function startEventDispatcher() {
        // Procesar eventos pendientes cada segundo
        $this->loop->addPeriodicTimer(1, function() {
            try {
                $stmt = $this->db->prepare("SELECT id, account_id, account_number, entity_type, entity_id, event_type, payload, created_at FROM webtrader_events WHERE processed = 0 ORDER BY id ASC LIMIT 100");
                $stmt->execute();
                $events = $stmt->fetchAll(PDO::FETCH_ASSOC);

                if (!$events || count($events) === 0) {
                    return;
                }

                foreach ($events as $event) {
                    $payload = null;
                    if (!empty($event['payload'])) {
                        try { $payload = json_decode($event['payload'], true); } catch (\Exception $e) { $payload = null; }
                    }

                    switch ($event['entity_type']) {
                        case 'account':
                            // Emitir actualización de cuenta
                            $data = [
                                'type' => 'account_update',
                                'data' => $payload ?: $this->getAccountDataByAccountId((int)$event['account_id'])
                            ];
                            $this->broadcast('account_update', $data);
                            break;
                        case 'order':
                            $data = [
                                'type' => 'order_update',
                                'data' => $payload ?: [
                                    'account_id' => (int)$event['account_id'],
                                    'order_id' => (int)$event['entity_id'],
                                    'event' => $event['event_type']
                                ]
                            ];
                            $this->broadcast('order_update', $data);
                            break;
                        case 'position':
                            $data = [
                                'type' => 'position_update',
                                'data' => $payload ?: [
                                    'account_id' => (int)$event['account_id'],
                                    'position_id' => (int)$event['entity_id'],
                                    'event' => $event['event_type']
                                ]
                            ];
                            $this->broadcast('position_update', $data);
                            break;
                        default:
                            // Otros tipos no se emiten por ahora
                            break;
                    }

                    // Marcar evento como procesado
                    $upd = $this->db->prepare("UPDATE webtrader_events SET processed = 1, processed_at = NOW() WHERE id = ?");
                    $upd->execute([$event['id']]);
                }
            } catch (\PDOException $e) {
                echo "❌ Error leyendo eventos realtime: " . $e->getMessage() . "\n";
            } catch (\Exception $e) {
                echo "❌ Error general en dispatcher: " . $e->getMessage() . "\n";
            }
        });
    }
    
    private function startPriceUpdates() {
        // Actualizar precios cada 1 segundo
        $this->loop->addPeriodicTimer(1, function() {
            $this->broadcastPriceUpdates();
        });

        // Heartbeat Finage cada 30s para mantener viva la conexión
        $this->loop->addPeriodicTimer(30, function() { $this->sendFinageHeartbeat(); });
        
        // Update market status every 30 seconds
        $this->loop->addPeriodicTimer(30, function() {
            $marketStatus = $this->finageAPI->getMarketStatus();
            
            $this->broadcast('market_status', [
                'type' => 'market_status',
                'data' => $marketStatus,
                'timestamp' => time()
            ]);
        });
        
        // Iniciar conexión WebSocket con Finage si está disponible
        $this->initializeFinageWebSocket();
    }

    /**
     * Obtener datos de cuenta por account_id
     */
    private function getAccountDataByAccountId($accountId) {
        try {
            $stmt = $this->db->prepare("SELECT account_number, balance, equity, margin, free_margin, margin_level, currency, leverage FROM trading_accounts WHERE id = ? LIMIT 1");
            $stmt->execute([$accountId]);
            return $stmt->fetch(\PDO::FETCH_ASSOC);
        } catch (\PDOException $e) {
            echo "❌ Error obteniendo datos de cuenta por ID: " . $e->getMessage() . "\n";
            return null;
        }
    }
    
    private function broadcastPriceUpdates() {
        // Obtener datos combinados de mercado (forex + stocks)
        try {
            $marketData = $this->finageAPI->getCombinedMarketData();
            
            // Broadcast forex data
            foreach ($marketData['forex'] as $symbol => $data) {
                $this->emitPriceAndUpdateCandles($symbol, $data);
            }
            
            // Broadcast stock data
            foreach ($marketData['stocks'] as $symbol => $data) {
                $this->emitPriceAndUpdateCandles($symbol, $data);
            }

            // Broadcast crypto data
            if (!empty($marketData['crypto'])) {
                foreach ($marketData['crypto'] as $symbol => $data) {
                    $this->emitPriceAndUpdateCandles($symbol, $data);
                }
            }
            
            // Broadcast market status
            $this->broadcast('market_status', [
                'type' => 'market_status',
                'data' => $marketData['market_status']
            ]);
            
        } catch (Exception $e) {
            echo "❌ Error en broadcast de precios: " . $e->getMessage() . "\n";
            // Sin fallback: mantener silencio hasta que el feed se restablezca
        }
    }
    
    /**
     * Broadcast a clientes suscritos a un símbolo específico
     */
    private function emitPriceAndUpdateCandles($symbol, $data) {
        // Normalizar estructura para frontend
        $payload = [
            'type' => 'price_update',
            'symbol' => $symbol,
            'bid' => $data['bid'] ?? null,
            'ask' => $data['ask'] ?? null,
            'timestamp' => $data['timestamp'] ?? round(microtime(true) * 1000),
            'change' => $data['change'] ?? null,
            'change_percent' => $data['change_percent'] ?? null,
            'source' => $data['source'] ?? 'finage',
        ];
        // Emitir a suscriptores por símbolo
        $this->broadcast("prices:{$symbol}", $payload);
        // También soportar clave legacy
        $this->broadcast("price_update:{$symbol}", $payload);
        // Actualizar agregadores de velas
        $bid = floatval($payload['bid']);
        $ask = floatval($payload['ask']);
        $ts = intval($payload['timestamp']);
        if ($bid > 0 && $ask > 0) {
            $this->updateCandleAggregators($symbol, $bid, $ask, $ts);
        }
    }
    
    // Eliminado: no se emitirán precios simulados
    
    /**
     * Inicializar conexión WebSocket con Finage para datos en tiempo real
     */
    private function initializeFinageWebSocket() {
        try {
            // Suscribirse a FOREX (pares mayormente solicitados) usando servidor dedicado
            $subs = $this->finageAPI->getWebSocketSubscriptions(true, false, false, false, false);
            if (empty($subs)) {
                echo "⚠️  No hay suscripciones WS configuradas; feed WS deshabilitado\n";
                return;
            }
            
            foreach ($subs as $wsConfig) {
                $wsUrl = $wsConfig['url'] . '?token=' . $wsConfig['token'];
                $symbolsList = implode(',', $wsConfig['symbols']);
                echo "🔗 Conectando a Finage WS (FOREX): {$wsUrl}\n";

                \Ratchet\Client\connect($wsUrl, [], [], $this->loop)->then(function (\Ratchet\Client\WebSocket $conn) use ($wsConfig, $symbolsList, $wsUrl) {
                    echo "✅ Conectado a WS (FOREX): {$wsConfig['url']}\n";
                    // Persistir conexión y símbolos iniciales
                    $this->finageWsConn = $conn;
                    $this->finageSubscribedForex = [];
                    foreach ($wsConfig['symbols'] as $s) { $this->finageSubscribedForex[$s] = true; }
                    // Avisar a clientes: feed LIVE conectado
                    $this->broadcast('feed_status', [
                        'type' => 'feed_status',
                        'feed' => 'finage',
                        'market' => 'forex',
                        'status' => 'connected'
                    ]);

                    if (!empty($symbolsList)) {
                        $subscribeMessage = json_encode([
                            'action' => 'subscribe',
                            'symbols' => $symbolsList
                        ]);
                        $conn->send($subscribeMessage);
                        echo "📡 Suscrito (FOREX): {$symbolsList}\n";
                    } else {
                        echo "ℹ️  Sin símbolos FOREX para suscripción en {$wsConfig['url']}\n";
                    }

                    $conn->on('message', function ($msg) {
                        // $msg es Ratchet\RFC6455\Messaging\MessageInterface
                        $payload = method_exists($msg, 'getPayload') ? $msg->getPayload() : (string)$msg;
                        $this->handleFinageWebSocketData($payload);
                    });
                    $conn->on('close', function ($code = null, $reason = null) use ($wsConfig) {
                        echo "⚠️  WS FOREX cerrado: {$wsConfig['url']} code=" . ($code ?? '-') . " reason=" . ($reason ?? '-') . "\n";
                        // Limpiar estado de conexión
                        $this->finageWsConn = null;
                        $this->finageSubscribedForex = [];
                        // Avisar a clientes: feed LIVE desconectado
                        $this->broadcast('feed_status', [
                            'type' => 'feed_status',
                            'feed' => 'finage',
                            'market' => 'forex',
                            'status' => 'disconnected'
                        ]);
                        // Intentar reconectar con backoff simple
                        $this->loop->addTimer(3, function() { $this->initializeFinageWebSocket(); });
                    });
                }, function ($error) use ($wsUrl) {
                    echo "❌ Error conectando a WS FOREX {$wsUrl}: " . $error->getMessage() . "\n";
                    // Avisar a clientes: error de feed
                    $this->broadcast('feed_status', [
                        'type' => 'feed_status',
                        'feed' => 'finage',
                        'market' => 'forex',
                        'status' => 'error',
                        'message' => $error->getMessage()
                    ]);
                    // Reintentar conexión
                    $this->loop->addTimer(5, function() { $this->initializeFinageWebSocket(); });
                });
            }
        } catch (Exception $e) {
            echo "❌ Error inicializando Finage WebSocket (FOREX): " . $e->getMessage() . "\n";
        }
    }

    // Suscripción dinámica a Finage WS para símbolos FOREX según demanda de clientes
    private function ensureFinageSubscription($symbol) {
        if (!$symbol || !$this->finageWsConn) return;
        if ($this->determineSymbolType($symbol) !== 'forex') return;
        if (isset($this->finageSubscribedForex[$symbol])) return;
        $msg = json_encode(['action' => 'subscribe', 'symbols' => $symbol]);
        try {
            $this->finageWsConn->send($msg);
            $this->finageSubscribedForex[$symbol] = true;
            echo "📡 Finage subscribe dinámico: {$symbol}\n";
        } catch (\Exception $e) {
            echo "❌ Error enviando subscribe a Finage para {$symbol}: {$e->getMessage()}\n";
        }
    }

    private function sendFinageHeartbeat() {
        if ($this->finageWsConn) {
            try { $this->finageWsConn->send(json_encode(['action' => 'ping'])); } catch (\Exception $e) {}
        }
    }

    // Intentar desuscribir símbolo FOREX en Finage si ya no hay clientes suscritos
    private function maybeUnsubscribeFinageSymbol($symbol) {
        if (!$symbol || !$this->finageWsConn) return;
        if ($this->determineSymbolType($symbol) !== 'forex') return;
        $channelKey = "prices:{$symbol}";
        if (!isset($this->subscriptions[$channelKey]) || empty($this->subscriptions[$channelKey])) {
            if (isset($this->finageSubscribedForex[$symbol])) {
                $msg = json_encode(['action' => 'unsubscribe', 'symbols' => $symbol]);
                try {
                    $this->finageWsConn->send($msg);
                    unset($this->finageSubscribedForex[$symbol]);
                    echo "🔕 Finage unsubscribe dinámico: {$symbol}\n";
                } catch (\Exception $e) {
                    echo "❌ Error enviando unsubscribe a Finage para {$symbol}: {$e->getMessage()}\n";
                }
            }
        }
    }
    
    /**
     * Manejar datos entrantes del WebSocket de Finage
     */
    private function handleFinageWebSocketData($data) {
        try {
            $messages = explode("\n", trim($data));
            
            foreach ($messages as $message) {
                if (empty($message)) continue;
                
                $priceData = json_decode($message, true);
                
                if ($priceData && isset($priceData['s'])) { // 's' = symbol
                    $formattedData = [
                        'symbol' => $priceData['s'],
                        'bid' => floatval($priceData['b'] ?? $priceData['bp'] ?? 0),
                        'ask' => floatval($priceData['a'] ?? $priceData['ap'] ?? 0),
                        'timestamp' => intval($priceData['t'] ?? time() * 1000),
                        'type' => $this->determineSymbolType($priceData['s']),
                        // Mapear campos de Finage WS
                        // dc: Daily change percentage, dd: Daily difference
                        'change_percent' => isset($priceData['dc']) ? floatval($priceData['dc']) : null,
                        'change' => isset($priceData['dd']) ? floatval($priceData['dd']) : null,
                        'source' => 'finage'
                    ];
                    
                    $formattedData['spread'] = $formattedData['ask'] - $formattedData['bid'];
                    // Emitir y actualizar velas
                    $this->emitPriceAndUpdateCandles($formattedData['symbol'], $formattedData);
                }
            }
            
        } catch (Exception $e) {
            echo "❌ Error procesando datos de Finage WebSocket: " . $e->getMessage() . "\n";
        }
    }

    /**
     * Actualizar agregadores de velas para todos los timeframes configurados
     */
    private function updateCandleAggregators($symbol, $bid, $ask, $timestampMs) {
        $mid = ($bid + $ask) / 2.0;
        if (!is_finite($mid)) return;
        foreach ($this->candleTimeframes as $tf) {
            $bucketMs = $tf * 60 * 1000;
            $bucketStart = intval(floor($timestampMs / $bucketMs) * $bucketMs);
            $key = "{$symbol}:{$tf}";
            $agg = $this->candleAggregators[$key] ?? null;
            if (!$agg || $agg['time'] !== $bucketStart) {
                // Finalizar vela anterior si existía
                if ($agg) {
                    $this->broadcastCandle($symbol, $tf, $agg, true);
                }
                // Iniciar nueva vela
                $agg = [
                    'time' => $bucketStart,
                    'open' => $mid,
                    'high' => $mid,
                    'low' => $mid,
                    'close' => $mid
                ];
            } else {
                // Actualizar vela en formación
                if ($mid > $agg['high']) $agg['high'] = $mid;
                if ($mid < $agg['low']) $agg['low'] = $mid;
                $agg['close'] = $mid;
            }
            $this->candleAggregators[$key] = $agg;
            // Emitir actualización parcial
            $this->broadcastCandle($symbol, $tf, $agg, false);
        }
    }

    private function broadcastCandle($symbol, $timeframe, $agg, $complete = false) {
        $payload = [
            'type' => 'candle_update',
            'symbol' => $symbol,
            'timeframe' => $timeframe,
            'candle' => [
                'time' => $agg['time'],
                'open' => $agg['open'],
                'high' => $agg['high'],
                'low' => $agg['low'],
                'close' => $agg['close'],
                'complete' => $complete
            ]
        ];
        $this->broadcast("candle_update:{$symbol}:{$timeframe}", $payload);
    }

    private function mapMinutesToTfCode($minutes) {
        // Mapear minutos a códigos Finage
        switch (intval($minutes)) {
            case 1: return 'M1';
            case 5: return 'M5';
            case 15: return 'M15';
            case 30: return 'M30';
            case 60: return 'H1';
            case 240: return 'H4';
            case 1440: return 'D1';
            default: return 'M1';
        }
    }

    private function fetchInitialCandles($symbol, $tfCode) {
        // Intentar obtener datos reales de Finage; sin fallback a demo
        try {
            $candles = $this->finageAPI->getIntradayData($symbol, $tfCode);
            // Convertir a estructura esperada por frontend
            return array_map(function($c) {
                return [
                    'x' => intval($c['time'] * 1000),
                    'o' => floatval($c['open']),
                    'h' => floatval($c['high']),
                    'l' => floatval($c['low']),
                    'c' => floatval($c['close'])
                ];
            }, $candles);
        } catch (\Exception $e) {
            // Sin datos: devolver arreglo vacío
            return [];
        }
    }
    
    /**
     * Determinar si un símbolo es forex o stock
     */
    private function determineSymbolType($symbol) {
        $config = include(__DIR__ . '/../../../config/finage_config.php');
        
        if (in_array($symbol, $config['default_symbols'])) {
            return 'forex';
        } else if (in_array($symbol, $config['stock_symbols'])) {
            return 'stock';
        } else if (in_array($symbol, $config['crypto_symbols'] ?? [])) {
            return 'crypto';
        }
        
        return 'unknown';
    }
}

// Iniciar servidor WebSocket
$server = new WebTraderWebSocketServer();
$server->start(8082);
