<?php

    /** @noinspection PhpMissingFieldTypeInspection */

    namespace TamerLib;

    use Exception;
    use InvalidArgumentException;
    use LogLib\Log;
    use RedisException;
    use RuntimeException;
    use TamerLib\Classes\JobManager;
    use TamerLib\Classes\RedisServer;
    use TamerLib\Classes\Utilities;
    use TamerLib\Classes\WorkerSupervisor;
    use TamerLib\Enums\JobStatus;
    use TamerLib\Enums\TamerMode;
    use TamerLib\Exceptions\ConnectionException;
    use TamerLib\Exceptions\JobManagerException;
    use TamerLib\Exceptions\JobNotFoundException;
    use TamerLib\Exceptions\ServerException;
    use TamerLib\Exceptions\TimeoutException;
    use TamerLib\Exceptions\WorkerFailedException;
    use TamerLib\Objects\JobPacket;
    use TamerLib\Objects\ServerConfiguration;
    use TamerLib\Objects\WorkerConfiguration;
    use Throwable;

    class tm
    {
        /**
         * @var string|null
         */
        private static $mode;

        /**
         * @var ServerConfiguration|null
         */
        private static $server_configuration;

        /**
         * @var RedisServer|null
         */
        private static $server;

        /**
         * @var int[]
         */
        private static $watching_jobs = [];

        /**
         * @var WorkerSupervisor|null
         */
        private static $supervisor;

        /**
         * @var JobManager|null
         */
        private static $job_manager;

        /**
         * @var WorkerConfiguration|null
         */
        private static $worker_configuration;

        /**
         * @var array
         */
        private static $function_pointers = [];

        /**
         * @var string|null
         */
        private static $return_channel;

        /**
         * INTERNAL FUNCTIONS
         */

        /**
         * Appends the job ID to the watch list
         *
         * @param int $job_id
         * @return void
         */
        private static function addToWatchlist(int $job_id): void
        {
            if(!in_array($job_id, self::$watching_jobs, true))
            {
                self::$watching_jobs[] = $job_id;
            }
        }

        /**
         * Removes the job ID from the watch list
         *
         * @param int $job_id
         * @return void
         */
        private static function removeFromWatchlist(int $job_id): void
        {
            if(($key = array_search($job_id, self::$watching_jobs, true)) !== false)
            {
                unset(self::$watching_jobs[$key]);
            }
        }

        /**
         * GLOBAL FUNCTIONS
         */

        /**
         * Initializes Tamer in the specified mode
         *
         * Note that Tamer can only be initialized once per process, additionally not all functions are available in
         * all modes. Please review the documentation for CLIENT & WORKER mode usage as both modes operate differently.
         *
         * CLIENT MODE:
         *  Client Mode supervises workers and optionally initializes a server instance if the $server_config parameter
         *  is left null, in which case a server will be initialized with default parameters. If a server is already
         *  initialized then CLIENT Mode can work in swarm mode where multiple clients can be initialized, and they will
         *  all share the same server instance. This is done by passing the same $server_config object to each client.
         *
         * WORKER MODE:
         *  Worker Mode is responsible for listening for jobs and executing them. Worker Mode can only be initialized
         *  if the parent process is a client process, otherwise an exception will be thrown because the worker will
         *  have no server to connect to if it is not initialized by a client.
         *
         * @param string|null $mode
         * @param ServerConfiguration|null $server_config
         * @return void
         * @throws ServerException
         * @throws Exception
         */
        public static function initialize(?string $mode, ?ServerConfiguration $server_config=null): void
        {
            if(self::$mode !== null)
            {
                throw new RuntimeException('TamerLib has already been initialized.');
            }

            if(!in_array(strtolower($mode), TamerMode::ALL, true))
            {
                throw new InvalidArgumentException(sprintf('Invalid mode "%s" provided, must be one of "%s".', $mode, implode('", "', TamerMode::ALL)));
            }

            self::$mode = $mode;
            self::$server_configuration = $server_config;

            if($server_config === null && $mode === TamerMode::CLIENT)
            {
                try
                {
                    // Initialize the server if no configuration was provided, and we are in client mode
                    self::$server_configuration = new ServerConfiguration();
                    self::$server = new RedisServer(self::$server_configuration);
                    self::$server->start();

                    // Register shutdown function to stop the server when the process exits
                    register_shutdown_function(static function()
                    {
                        self::$server?->stop();
                    });
                }
                catch(Exception $e)
                {
                    throw new ServerException('Failed to initialize the server.', 0, $e);
                }

            }

            if($mode === TamerMode::WORKER)
            {
                try
                {
                    self::$worker_configuration = WorkerConfiguration::fromEnvironment();
                    self::$server_configuration = new ServerConfiguration(self::$worker_configuration->getHost(), self::$worker_configuration->getPort(), self::$worker_configuration->getPassword());
                }
                catch(Exception $e)
                {
                    throw new RuntimeException('Failed to initialize worker configuration. (is the process running as a worker?)', 0, $e);
                }
            }

            if($mode === TamerMode::CLIENT)
            {
                self::$supervisor = new WorkerSupervisor(self::$server_configuration);
                self::$return_channel = sprintf('rch%s', random_int(100000000, 999999999));
            }

            self::$job_manager = new JobManager(self::$server_configuration);
        }

        /**
         * Shuts down all workers
         *
         * @return void
         * @noinspection PhpUnused
         */
        public static function shutdown(): void
        {
            if(self::$mode === null)
            {
                return;
            }

            if(self::$mode === TamerMode::CLIENT && self::$supervisor !== null)
            {
                self::$supervisor->stopAll();
            }
        }

        /**
         * Returns the current mode Tamer is running in
         *
         * @return string
         */
        public static function getMode(): string
        {
            return self::$mode ?? TamerMode::NONE;
        }

        /**
         * Monitors all internal processes
         *
         * @param int $timeout
         * @return void
         */
        public static function monitor(int $timeout=0): void
        {
            if($timeout > 0)
            {
                try
                {
                    self::$supervisor?->monitor($timeout);
                    self::$server?->monitor($timeout);
                }
                catch(Exception $e)
                {
                    Log::error(Utilities::getName(), $e->getMessage(), $e);
                }
            }
            else
            {
                $start_time = time();
                while(true)
                {
                    try
                    {
                        self::$supervisor?->monitor(-1);
                        self::$server?->monitor(-1);
                    }
                    catch(Exception $e)
                    {
                        Log::error(Utilities::getName(), $e->getMessage(), $e);
                    }

                    if(time() - $start_time >= $timeout)
                    {
                        break;
                    }
                }
            }
        }

        /**
         * CLIENT FUNCTIONS
         */

        /**
         * Spawns a worker process by their count, if the path is null then a generic sub process will be spawned
         * that will only be capable of executing closures.
         *
         * @param int $count
         * @param string|null $path
         * @param int $channel
         * @return void
         * @throws WorkerFailedException
         */
        public static function createWorker(int $count=8, ?string $path=null, int $channel=0): void
        {
            if(self::$mode !== TamerMode::CLIENT)
            {
                throw new RuntimeException(sprintf('Attempting to spawn a worker in \'%s\' mode, only clients can spawn workers.', self::$mode));
            }

            if($path === null)
            {
                self::$supervisor->spawnClosure($count, $channel);
            }
            else
            {
                self::$supervisor->spawnWorker($path, $count, $channel);
            }

            self::monitor(-1);
        }

        /**
         * Preforms a function call against a worker in the background, returns the Job ID to keep track of the job status.
         *
         * @param string $function
         * @param array $arguments
         * @param int $channel
         * @return mixed
         */
        public static function do(string $function, array $arguments, int $channel=0): mixed
        {
            if(self::$mode !== TamerMode::CLIENT)
            {
                throw new RuntimeException(sprintf('Attempting to do() in \'%s\' mode, only clients can preform do().', self::$mode));
            }

            $job_packet = new JobPacket();
            $job_packet->setParameters(serialize($arguments));
            $job_packet->setPayload($function);
            $job_packet->setChannel($channel);
            $job_packet->setReturnChannel(self::$return_channel);

            try
            {
                self::$job_manager->pushJob($job_packet);
            }
            catcH(Exception $e)
            {
                throw new RuntimeException('do() failed, failed to push job to the server', 0, $e);
            }

            self::addToWatchlist($job_packet->getId());
            return $job_packet->getId();
        }

        /**
         * Sends a function call to a worker in the background, but once the job is completed it will be forgotten and
         * the result will not be returned, this also means that the job will not be added to the watchlist.
         *
         * @param string $function
         * @param array $arguments
         * @param int $channel
         * @return void
         * @noinspection PhpUnused
         */
        public static function dof(string $function, array $arguments, int $channel=0): void
        {
            if(self::$mode !== TamerMode::CLIENT)
            {
                throw new RuntimeException(sprintf('Attempting to dof() in \'%s\' mode, only clients can preform dof().', self::$mode));
            }

            $job_packet = new JobPacket();
            $job_packet->setParameters(serialize($arguments));
            $job_packet->setPayload($function);
            $job_packet->setChannel($channel);

            try
            {
                self::$job_manager->pushJob($job_packet);
            }
            catcH(Exception $e)
            {
                throw new RuntimeException('dof() failed, failed to push job to the server', 0, $e);
            }

            self::addToWatchlist($job_packet->getId());
        }

        /**
         * Preforms a do() call on a waitFor() call all in one function.
         *
         * @param string $function
         * @param array $arguments
         * @param int $channel
         * @param int $timeout
         * @return mixed
         * @throws JobNotFoundException
         * @throws ServerException
         * @throws Throwable
         * @throws TimeoutException
         * @noinspection PhpUnused
         */
        public static function doWait(string $function, array $arguments, int $channel=0, int $timeout=0): mixed
        {
            return self::waitFor(self::do($function, $arguments, $channel), $timeout);
        }

        /**
         * Waits for all the dispatched jobs to complete, this is a blocking function and will not return until all the
         * jobs have completed. If a timeout is specified, the function will return after the timeout has been reached.
         *
         * @param callable $callback
         * @param int $timeout
         * @return void
         * @throws ServerException
         * @throws Throwable
         * @throws TimeoutException
         */
        public static function wait(callable $callback, int $timeout=0): void
        {
            if(self::$mode !== TamerMode::CLIENT)
            {
                throw new RuntimeException(sprintf('Attempting to wait() in \'%s\' mode, only clients can preform wait().', self::$mode));
            }

            $time_start = time();
            while(true)
            {
                self::monitor(-1);
                if(count(self::$watching_jobs) === 0)
                {
                    Log::debug(Utilities::getName(), 'No jobs to wait for, returning');
                    return;
                }

                Log::debug(Utilities::getName(), 'Waiting for jobs to complete');
                $job_packet = self::$job_manager->listenReturnChannel(self::$return_channel);

                if(in_array($job_packet->getId(), self::$watching_jobs))
                {
                    self::removeFromWatchlist($job_packet->getId());
                    self::$job_manager->dropJob($job_packet->getId());

                    if($job_packet->getStatus() === JobStatus::FINISHED)
                    {
                        $return_value = $job_packet->getReturnValue();

                        if($return_value === null)
                        {
                            $return_value = null;
                        }
                        else
                        {
                            $return_value = unserialize($return_value, ['allowed_classes' => true]);

                            if($return_value === false)
                            {
                                Log::error(Utilities::getName(), 'Failed to unserialize return value, return value was dropped');
                                $return_value = null;
                            }
                        }

                        $callback($job_packet->getId(), $return_value);
                    }
                    elseif($job_packet->getStatus() === JobStatus::FAILED)
                    {
                        try
                        {
                            $e = unserialize($job_packet->getException(), ['allowed_classes' => true]);
                        }
                        catch(Exception $e)
                        {
                            Log::error(Utilities::getName(), 'Failed to unserialize exception, exception was dropped', $e);
                        }
                        finally
                        {
                            if(isset($e) && $e instanceof Throwable)
                            {
                                throw $e;
                            }

                            throw new ServerException('wait() failed, job returned with an exception');
                        }
                    }
                    else
                    {
                        Log::debug(Utilities::getName(), sprintf('Job \'%s\' returned with an unexpected status of \'%s\'', $job_packet->getId(), $job_packet->getStatus()));
                        throw new ServerException('wait() failed, job returned with an unexpected status of \'' . $job_packet->getStatus() . '\'');
                    }
                }
                else
                {
                    Log::debug(Utilities::getName(), sprintf('Job \'%s\' has returned, but is not in the watchlist', $job_packet->getId()));
                }

                if ($timeout < 0)
                {
                    throw new TimeoutException('wait() timed out');
                }

                if($timeout > 0 && (time() - $time_start) >= $timeout)
                {
                    throw new TimeoutException('wait() timed out');
                }

                usleep(10);
            }
        }

        /**
         * Waits for a job to complete, returns the result of the job.
         *
         * @param JobPacket|int $job_id
         * @param int $timeout
         * @return mixed
         * @throws JobNotFoundException
         * @throws ServerException
         * @throws TimeoutException
         * @throws Throwable
         */
        public static function waitFor(JobPacket|int $job_id, int $timeout=0): mixed
        {
            if(self::$mode !== TamerMode::CLIENT)
            {
                throw new RuntimeException(sprintf('Attempting to waitFor() in \'%s\' mode, only clients can preform waitFor().', self::$mode));
            }

            if($job_id instanceof JobPacket)
            {
                $job_id = $job_id->getId();
            }

            $time_start = time();

            while(true)
            {
                self::monitor(-1);
                switch(self::$job_manager->getJobStatus($job_id))
                {
                    case JobStatus::FINISHED:
                        $return = self::$job_manager->getJobResult($job_id);
                        self::$job_manager->dropJob($job_id);
                        return $return;

                    case JobStatus::FAILED:
                        $throwable = self::$job_manager->getJobException($job_id);
                        self::$job_manager->dropJob($job_id);
                        throw $throwable;
                }

                if($timeout < 0)
                {
                    throw new TimeoutException('waitFor() timed out');
                }

                if($timeout > 0 && (time() - $time_start) >= $timeout)
                {
                    throw new TimeoutException(sprintf('waitFor() timed out after %d seconds', $timeout));
                }

                usleep(10);
            }
        }


        /**
         * Clears the watchlist, this will remove all jobs from the watchlist.
         *
         * @return void
         * @noinspection PhpUnused
         */
        public static function clear(): void
        {
            if(self::$mode !== TamerMode::CLIENT)
            {
                throw new RuntimeException(sprintf('Attempting to clear() in \'%s\' mode, only clients can preform clear().', self::$mode));
            }

            self::$watching_jobs = [];
        }

        /**
         * Invokes the call() function, returns the Job ID.
         *
         * @param string $name
         * @param array $arguments
         * @return mixed
         */
        public static function __callStatic(string $name, array $arguments)
        {
            return self::do($name, $arguments);
        }

        /**
         * WORKER FUNCTIONS
         */

        /**
         * @param string $function
         * @param callable $callback
         * @return void
         */
        public static function addFunction(string $function, callable $callback): void
        {
            if(self::$mode !== TamerMode::WORKER)
            {
                throw new RuntimeException(sprintf('Attempting to addFunction() in \'%s\' mode, only workers can preform addFunction().', self::$mode));
            }

            self::$function_pointers[$function] = $callback;
        }

        /**
         * @param string $function
         * @return void
         * @noinspection PhpUnused
         */
        public static function removeFunction(string $function): void
        {
            if(self::$mode !== TamerMode::WORKER)
            {
                throw new RuntimeException(sprintf('Attempting to removeFunction() in \'%s\' mode, only workers can preform removeFunction().', self::$mode));
            }

            unset(self::$function_pointers[$function]);
        }

        /**
         * @return array
         * @noinspection PhpUnused
         */
        public static function getFunctions(): array
        {
            if(self::$mode !== TamerMode::WORKER)
            {
                throw new RuntimeException(sprintf('Attempting to getFunctions() in \'%s\' mode, only workers can preform getFunctions().', self::$mode));
            }

            return array_keys(self::$function_pointers);
        }

        /**
         * @param int|array $channel
         * @param int $timeout
         * @return void
         * @throws ConnectionException
         * @throws JobManagerException
         * @throws JobNotFoundException
         * @throws ServerException
         */
        public static function run(int|array $channel=0, int $timeout=0): void
        {
            if(self::$mode !== TamerMode::WORKER)
            {
                throw new RuntimeException(sprintf('Attempting to run() in \'%s\' mode, only workers can preform run().', self::$mode));
            }

            $start_time = time();
            $error_time = null;

            while(true)
            {
                try
                {
                    $job_packet = self::$job_manager->listenForJob(self::$worker_configuration->getWorkerId(), $channel, $timeout);
                    break;
                }
                catch(TimeoutException $e)
                {
                    unset($e);
                    return;
                }
                /** @noinspection PhpRedundantCatchClauseInspection */
                catch(RedisException $e)
                {
                    // TODO: There has to be a better way to do this.
                    if(strtolower($e->getMessage()) === 'redis server went away')
                    {
                        if($error_time === null)
                        {
                            $error_time = time();
                        }
                        else if((time() - $error_time) >= 5)
                        {
                            throw new ServerException('Redis server went away, and did not come back.');
                        }
                    }
                }

                if((time() - $start_time) >= $timeout)
                {
                    return;
                }
            }


            Log::debug(Utilities::getName(), sprintf('Worker %s received job %s', self::$worker_configuration->getWorkerId(), $job_packet->getId()));

            if(!isset(self::$function_pointers[$job_packet->getPayload()]))
            {

                Log::warning(Utilities::getName(), sprintf('Job %s requested function \'%s\' which does not exist, rejecting job.', $job_packet->getId(), $job_packet->getPayload()));
                self::$job_manager->rejectJob($job_packet);
            }

            try
            {
                $result = call_user_func_array(self::$function_pointers[$job_packet->getPayload()], unserialize($job_packet->getParameters(), ['allowed_classes'=>true]));
                self::$job_manager->returnJob($job_packet, $result);
            }
            catch(Exception $e)
            {
                self::$job_manager->returnException($job_packet, $e);
            }
        }
    }