<?php

namespace ReactphpX\Bridge;

use Evenement\EventEmitterTrait;
use React\Promise\Deferred;
use function React\Async\async;
use React\Promise\Timer\TimeoutException;
use Ramsey\Uuid\Uuid;
use React\Stream\DuplexStreamInterface;
use ReactphpX\Bridge\Interface\ServerInterface;
use ReactphpX\Bridge\Interface\CallInterface;
use ReactphpX\Bridge\Info;
use ReactphpX\Bridge\Interface\VerifyInterface;
use React\EventLoop\Loop;

class Server implements ServerInterface
{
    use EventEmitterTrait;

    public static $debug = false;

    protected $call;
    protected $verify;
    protected $clients;
    protected $controllerConnections;
    protected $tunnelConnections;
    protected $tmpConnections;
    protected $uuidToControlerConnections = [];

    protected $uuidToDeferred = [];

    public function __construct(VerifyInterface $verify)
    {
        $this->verify = $verify;

        // $this->uuids = $uuids;
        $this->clients = new \SplObjectStorage;

        $this->tmpConnections = new \SplObjectStorage;
        $this->controllerConnections = new \SplObjectStorage;
        $this->tunnelConnections = new \SplObjectStorage;
    }

    public function setCall(CallInterface $call)
    {
        $this->call = $call;
    }

    public function call($closure, $params = null, $data = [])
    {
        return $this->call->call($closure, $params, $data);
    }

    public function onOpen(DuplexStreamInterface $stream, $info = null)
    {
        if (!isset($info['decodeEncodeClass'])) {
            // error
            throw new \InvalidArgumentException('decodeEncodeClass is required');
        }

        $decodeEncode = new $info['decodeEncodeClass'];

        $stream->on('data', function ($buffer) use ($stream) {
            if (self::$debug) {
                echo "=====> onMessage <========" . "\n";
                echo $buffer . "\n";
            }
        });

        $stream->write($decodeEncode->encode([
            'cmd' => 'init',
            'uuid' => Uuid::uuid4()->toString(),
            'data' => [
                'remote_address' => $info['remote_address'] ?? '',
            ]
        ]));


        $hash = spl_object_hash($stream);
        echo "New connection! {$hash}\n";
        $this->clients->attach($stream, new Info($info + [
            'decodeEncode' => $decodeEncode,
            'active_time' => time(),
        ]));
        $this->tmpConnections->attach($stream);
        echo "connection count({$this->clients->count()})\n";
    }

    public function onMessage(DuplexStreamInterface $stream, $msg)
    {
        // tmp data
        if ($this->tmpConnections->contains($stream)) {
            if ($messages = $this->clients[$stream]['decodeEncode']->decode($msg)) {
                foreach ($messages as $message) {
                    $this->handleTmpData($stream, $message);
                }
            }
        }
        // control data
        else if ($this->controllerConnections->contains($stream)) {
            if ($messages = $this->clients[$stream]['decodeEncode']->decode($msg)) {
                foreach ($messages as $message) {
                    $this->handleControlData($stream, $message);
                }
            }
        }
        // tunnel data pool to handle
        else if ($this->tunnelConnections->contains($stream)) {
            $this->tunnelConnections[$stream]['request_number'] = $this->tunnelConnections[$stream]['request_number'] + 1;
            $control = $this->tunnelConnections[$stream]['control'];
            if ($this->controllerConnections->contains($control)) {
                $this->controllerConnections[$control]['request_number'] = $this->controllerConnections[$control]['request_number'] + 1;
                $this->controllerConnections[$control]['tunnelConnections'][$stream]['request_number'] = $this->controllerConnections[$control]['tunnelConnections'][$stream]['request_number'] + 1;
            }
        }

        if ($this->clients->contains($stream)) {
            $this->clients[$stream]['active_time'] = time();
        }
    }

    protected function handleTmpData($stream, $message)
    {
        if ($this->controllerConnections->contains($stream)) {
            $this->handleControlData($stream, $message);
            return;
        }

        if (!is_array($message)) {
            $stream->close();
            return;
        }
        $uuid = $message['uuid'] ?? null;
        $cmd = $message['cmd'] ?? null;
        // 注册控制器
        if ($cmd == 'registerController') {
            echo "$uuid registerController\n";
            if (!$this->verify->verify($uuid)) {
                $stream->close();
                echo "$uuid registerController fail\n";
                return;
            }

            if (isset($this->uuidToControlerConnections[$uuid]) && $this->uuidToControlerConnections[$uuid]->count() >= 2) {
                echo "$uuid registerController fail over 2\n";
                $stream->close();
                return;
            }
            if ($this->controllerConnections->contains($stream)) {
                echo "$uuid registerController fail already register\n";
                $stream->close();
                return;
            }
            $this->controllerConnections->attach($stream, new Info([
                'request_number' => 0,
                'uuid' => $uuid,
                'tunnelConnections' => new \SplObjectStorage,
            ]));
            $this->tmpConnections->detach($stream);
            if (!isset($this->uuidToControlerConnections[$uuid])) {
                $this->uuidToControlerConnections[$uuid] = new \SplObjectStorage;
            }
            $this->uuidToControlerConnections[$uuid]->attach($stream);

            $this->emit($uuid . '_controllerConnected', [$stream]);
            $stream->write($this->clients[$stream]['decodeEncode']->encode([
                'cmd' => 'controllerConnected',
                'uuid' => $uuid,
                'data' => [
                    'something' => $this->verify->getSomethingByUuid($uuid),
                ]
            ]));
            // ping
            echo "$uuid registerController success\n";
        } else if ($cmd == 'registerTunnel') {
            if (isset($this->uuidToDeferred[$uuid])) {
                $this->tmpConnections->detach($stream);
                $deferred = $this->uuidToDeferred[$uuid];
                $deferred->resolve($stream);
                echo "registerTunnel success\n";
            } else {
                $stream->close();
            }
        } else {
            echo "unknown tmp cmd $cmd\n";
            $stream->close();
        }
    }

    protected function handleControlData($controlStream, $message)
    {
        if (!is_array($message)) {
            $controlStream->close();
            return;
        }

        $uuid = $message['uuid'] ?? null;
        $cmd = $message['cmd'] ?? null;
        if ($cmd == 'ping') {
            $controlStream->write($this->clients[$controlStream]['decodeEncode']->encode([
                'cmd' => 'pong',
                'uuid' => $uuid,
            ]));
        } else if ($cmd == 'pong') {
            if (isset($this->uuidToDeferred[$uuid])) {
                $this->uuidToDeferred[$uuid]->resolve(true);
            }
        } else if ($cmd === 'callback_peer_stream') {
            async(function () use ($uuid, $message) {
                try {

                    $peerUuid = $message['data']['params']['uuid'] ?? null;
                    if (!$peerUuid) {
                        $something = $message['data']['params']['something'] ?? null;
                        $peerUuid = $this->verify->getUuidBySomething($something);
                    }
                    if (!$peerUuid || !$this->verify->verify($peerUuid)) {
                        echo "peerUuid verify fail\n";
                        return;
                    }

                    $stream = $this->call->call($message['data']['self_serialized'], [
                        'uuid' => $uuid,
                    ], [
                        'event' => $message['data']['event'] ?? ''
                    ]);

                    $data = '';
                    $stream->on('data', $fn = function ($buffer) use (&$data) {
                        $data .= $buffer;
                    });

                    $peerStream = $this->call->call($message['data']['peer_serialized'] ?? '', [
                        'uuid' => $peerUuid,
                    ], $message['data']['data'] ?? []);

                    if ($data) {
                        $peerStream->write($data);
                        $data = '';
                    }

                    $stream->removeListener('data', $fn);
                    $fn = null;

                    $stream->pipe($peerStream, [
                        'end' => true
                    ]);

                    $peerStream->pipe($stream, [
                        'end' => true
                    ]);
                    $peerStream->on('close', function () use ($stream) {
                        $stream->end();
                    });

                    $stream->on('close', function () use ($peerStream) {
                        $peerStream->end();
                    });

                    $peerStream->on('error', function ($e) use ($stream) {
                        $stream->emit('error', [$e]);
                    });

                    $stream->on('error', function ($e) use ($peerStream) {
                        echo "stream error {$e->getMessage()}\n";
                    });
                } catch (\Throwable $th) {
                    echo $th->getFile() . "\n";
                    echo $th->getLine() . "\n";
                    echo $th->getMessage() . "\n";
                }
            })();
        } else if ($cmd === 'extend_cmd') {
            $this->emit('extend_cmd', [$message, $controlStream]);
        }
    }

    public function getControlUuidByTunnelStream($tunnelStream)
    {
        if ($this->tunnelConnections->contains($tunnelStream)) {
            return $this->tunnelConnections[$tunnelStream]['control_uuid'];
        }
        return null;
    }

    public function onClose(DuplexStreamInterface $stream, $reason = null)
    {
        echo "Server onClose\n";
        if ($this->tmpConnections->contains($stream)) {
            $this->tmpConnections->detach($stream);
        } else if ($this->controllerConnections->contains($stream)) {
            echo "controllerConnections onClose\n";
            $uuid = $this->controllerConnections[$stream]['uuid'];
            $this->controllerConnections->detach($stream);
            $this->uuidToControlerConnections[$uuid]->detach($stream);
            if ($this->uuidToControlerConnections[$uuid]->count() == 0) {
                unset($this->uuidToControlerConnections[$uuid]);
            }
        } else if ($this->tunnelConnections->contains($stream)) {
            echo "tunnelConnection onClose\n";
            $control = $this->tunnelConnections[$stream]['control'];
            if ($this->controllerConnections->contains($control)) {
                $this->controllerConnections[$control]['tunnelConnections']->detach($stream);
            }
            $this->tunnelConnections->detach($stream);
        }

        if ($this->clients->contains($stream)) {
            $this->clients->detach($stream);
        }
        echo "connection count({$this->clients->count()})\n";
    }

    public function onError(DuplexStreamInterface $stream, \Exception $e)
    {
        echo "Server onError\n";
        echo "An error has occurred: {$e->getMessage()}\n";
        echo $e->getTraceAsString() . "\n";
        $stream->close();
    }

    public function createConnection($params = null, $timeout = 3)
    {
        echo "createConnection-1\n";
        $deferred = new Deferred;
        $tunnelUuid = Uuid::uuid4()->toString();
        $this->uuidToDeferred[$tunnelUuid] = $deferred;

        $control = null;
        $event = null;
        $fn = function ($controlConnection) use ($tunnelUuid, &$control) {
            $control = $controlConnection;
            $controlConnection->write($this->clients[$control]['decodeEncode']->encode([
                'cmd' => 'createTunnelConnection',
                'uuid' => $tunnelUuid
            ]));
        };

        $uuid = $params['uuid'] ?? null;
        if ($uuid) {
            $uuids = explode(',', $uuid);
            $uuid = array_shift($uuids);
            if (isset($this->uuidToControlerConnections[$uuid]) && $this->uuidToControlerConnections[$uuid]->count() > 0) {
                $fn($this->getControllerConnection($this->uuidToControlerConnections[$uuid]));
                $fn = null;
            } else {
                $event = $uuid . '_controllerConnected';
                $this->once($event, $fn);
            }
        } else {
            $deferred->reject(new \RuntimeException('uuid is required'));
        }

        return \React\Promise\Timer\timeout($deferred->promise(), $timeout)->then(function ($connection) use ($tunnelUuid, &$control) {
            unset($this->uuidToDeferred[$tunnelUuid]);
            $this->tunnelConnections->attach($connection, new Info([
                'control_uuid' => $this->controllerConnections[$control]['uuid'],
                'control' => $control,
                'request_number' => 0,
            ]));
            $this->controllerConnections[$control]['tunnelConnections']->attach($connection, new Info([
                'request_number' => 0,
            ]));
            return [$connection, $this->clients[$connection]['decodeEncode']];
        }, function ($e) use ($tunnelUuid, $fn, $event, $uuids) {
            if ($fn) {
                $this->removeListener($event, $fn);
            }
            unset($this->uuidToDeferred[$tunnelUuid]);
            if ($e instanceof TimeoutException) {
                // continue try next client
                if (!empty($uuids)) {
                    return $this->createConnection([
                        'uuid' => implode(',', $uuids)
                    ], $e->getTimeout());
                }
                throw new \RuntimeException(
                    'wait timed out after ' . $e->getTimeout() . ' seconds (ETIMEDOUT)',
                    \defined('SOCKET_ETIMEDOUT') ? \SOCKET_ETIMEDOUT : 110
                );
            }
            throw $e;
        });
    }

    public function getConnections()
    {
        return $this->controllerConnections;
    }

    private function getControllerConnection($connections)
    {
        $currentConnection = null;
        foreach ($connections as $connection) {
            if (!$currentConnection) {
                $currentConnection = $connection;
            } else {
                if ($this->controllerConnections[$connection]['request_number'] < $this->controllerConnections[$currentConnection]['request_number']) {
                    $currentConnection = $connection;
                }
            }
        }
        return $currentConnection;
    }

    public function enableKeepAlive($interval = 30)
    {
        Loop::addPeriodicTimer($interval, function () use ($interval) {
            foreach ($this->clients as $client) {
                if ($this->tunnelConnections->contains($client)) {
                    // tunnel connection ping
                    $info = $this->clients[$client];
                    // 空闲去 ping
                    if ((time() - $info['active_time']) > $interval) {
                        if (method_exists($this->call, '_ping')) {
                            $this->call->_ping($client)->then(function(){
                                echo "tunnel ping success\n";
                            }, function ($e) {
                                echo "tunnel ping fail\n";
                                echo $e->getMessage() . "\n";
                            });
                        }
                    }
                } else {
                    $info = $this->clients[$client];
                    if ((time() - $info['active_time']) > $interval) {
                        $this->ping($client, $info['decodeEncode'])->then(null, function ($e) use ($client) {
                            echo "client ping fail\n";
                            echo $e->getMessage() . "\n";
                            $client->close();
                        });
                    }
                }
            }
        });
    }

    protected function ping($conn, $decodeEncode)
    {
        $deferred = new Deferred;
        $uuid = Uuid::uuid4()->toString();
        $conn->write($decodeEncode->encode([
            'cmd' => 'ping',
            'uuid' => $uuid,
        ]));
        $this->uuidToDeferred[$uuid] = $deferred;
        return \React\Promise\Timer\timeout($deferred->promise(), 3)->then(function () use ($uuid) {
            unset($this->uuidToDeferred[$uuid]);
        }, function ($e) use ($uuid) {
            unset($this->uuidToDeferred[$uuid]);
            if ($e instanceof TimeoutException) {
                throw new \RuntimeException(
                    'Connection timed out'
                );
            }
            throw $e;
        });
    }
}