import asyncio
import logging
import threading

from abc import ABC, abstractmethod
from functools import wraps
from threading import RLock
from typing import Any, Callable, Dict, Optional, Union

from platypush.bus import Bus
from platypush.common import ExtensionWithManifest
from platypush.event import EventGenerator
from platypush.message.response import Response
from platypush.utils import get_decorators, get_plugin_name_by_class

from ._actions import register_action, unregister_action

PLUGIN_STOP_TIMEOUT = 5  # Plugin stop timeout in seconds
_logger = logging.getLogger(__name__)


def action(f: Callable[..., Any]) -> Callable[..., Response]:
    """
    Decorator used to wrap the methods in the plugin classes that should be
    exposed as actions.

    It wraps the method's response into a generic
    :meth:`platypush.message.response.Response` object.
    """

    @wraps(f)
    def _execute_action(*args, **kwargs) -> Response:
        response = Response()
        action_id = None

        try:
            action_id = register_action(
                {
                    'action': '.'.join(
                        [get_plugin_name_by_class(args[0].__class__), f.__name__]
                    ),
                    'args': kwargs,
                }
            )
            result = f(*args, **kwargs)
        except Exception as e:
            if isinstance(e, KeyboardInterrupt):
                return response

            _logger.exception(e)
            result = Response(errors=[str(e)])

        if result and isinstance(result, Response):
            result.errors = (
                result.errors if isinstance(result.errors, list) else [result.errors]
            )
            response = result
        elif isinstance(result, tuple) and len(result) == 2:
            response.errors = result[1] if isinstance(result[1], list) else [result[1]]

            if len(response.errors) == 1 and response.errors[0] is None:
                response.errors = []
            response.output = result[0]
        else:
            response = Response(output=result, errors=[])

        unregister_action(action_id, response=response)
        return response

    # Propagate the docstring
    _execute_action.__doc__ = f.__doc__
    # Expose the wrapped function
    _execute_action.wrapped = f  # type: ignore
    return _execute_action


class Plugin(EventGenerator, ExtensionWithManifest):  # lgtm [py/missing-call-to-init]
    """Base plugin class"""

    def __init__(self, **kwargs):
        super().__init__()

        self.logger = logging.getLogger(
            'platypush:plugin:' + get_plugin_name_by_class(self.__class__)
        )
        if 'logging' in kwargs:
            self.logger.setLevel(getattr(logging, kwargs['logging'].upper()))

        self.registered_actions = set(
            get_decorators(self.__class__, climb_class_hierarchy=True).get('action', [])
        )

        PluginRegistry.get().register(self)

    @property
    def _db(self):
        """
        :return: The reference to the :class:`platypush.plugins.db.DbPlugin`.
        """
        from platypush.context import get_plugin
        from platypush.plugins.db import DbPlugin

        db: DbPlugin = get_plugin(DbPlugin)  # type: ignore
        if not (db):
            raise AssertionError('db plugin not initialized')
        return db

    @property
    def _redis(self):
        """
        :return: The reference to the :class:`platypush.plugins.redis.RedisPlugin`.
        """
        from platypush.context import get_plugin
        from platypush.plugins.redis import RedisPlugin

        redis: RedisPlugin = get_plugin(RedisPlugin)  # type: ignore
        if not (redis):
            raise AssertionError('db plugin not initialized')
        return redis

    @property
    def _bus(self):
        """
        :return: The reference to the :class:`platypush.bus.Bus`.
        """
        from platypush.context import get_bus

        return get_bus()

    @property
    def _entities(self):
        """
        :return: The reference to the :class:`platypush.plugins.entities.EntitiesPlugin`.
        """
        from platypush.context import get_plugin
        from platypush.plugins.entities import EntitiesPlugin

        entities: EntitiesPlugin = get_plugin('entities')  # type: ignore
        if not (entities):
            raise AssertionError('entities plugin not initialized')
        return entities

    @property
    def _plugin_name(self):
        return get_plugin_name_by_class(self.__class__)

    def __str__(self):
        """
        :return: The qualified name of the plugin.
        """
        return get_plugin_name_by_class(self.__class__)

    def run(self, method, *args, **kwargs):
        if not (method in self.registered_actions):
            raise AssertionError(
                f'{method} is not a registered action on {self.__class__.__name__}'
            )
        return getattr(self, method)(*args, **kwargs)


class RunnablePlugin(Plugin):
    """
    Class for runnable plugins - i.e. plugins that have a start/stop method and can be started.
    """

    def __init__(
        self,
        poll_interval: Optional[float] = 15,
        stop_timeout: Optional[float] = PLUGIN_STOP_TIMEOUT,
        disable_monitor: bool = False,
        **kwargs,
    ):
        """
        :param poll_interval: How often the :meth:`.loop` function should be
            executed (default: 15 seconds). *NOTE*: For back-compatibility
            reasons, the `poll_seconds` argument is also supported, but it's
            deprecated.
        :param stop_timeout: How long we should wait for any running
            threads/processes to stop before exiting (default: 5 seconds).
        :param disable_monitor: If set to True then the plugin will not monitor
            for new events. This is useful if you want to run a plugin in
            stateless mode and only leverage its actions, without triggering any
            events. Defaults to False.
        """
        super().__init__(**kwargs)
        self.poll_interval = poll_interval
        self.bus: Optional[Bus] = None
        self.disable_monitor = disable_monitor
        self._should_stop = threading.Event()
        self._stop_timeout = stop_timeout
        self._thread: Optional[threading.Thread] = None

        if kwargs.get('poll_seconds') is not None:
            self.logger.warning('poll_seconds is deprecated, use poll_interval instead')

            if self.poll_interval is None:
                self.poll_interval = kwargs['poll_seconds']

    def main(self):
        """
        Implementation of the main loop of the plugin.

        :meta private:
        """
        raise NotImplementedError()

    def should_stop(self) -> bool:
        return self._should_stop.is_set()

    def wait_stop(self, timeout=None):
        """
        Wait until a stop event is received.

        :meta private:
        """
        if self.disable_monitor:
            # Wait indefinitely if the monitor is disabled
            return self._should_stop.wait(timeout=None)

        return self._should_stop.wait(timeout=timeout)

    def start(self):
        """
        Start the plugin.

        :meta private:
        """
        self._thread = threading.Thread(
            target=self._runner, name=self.__class__.__name__, daemon=True
        )
        self._thread.start()

    def stop(self):
        """
        Stop the plugin.

        :meta private:
        """
        self._should_stop.set()
        if (
            self._thread
            and self._thread != threading.current_thread()
            and self._thread.is_alive()
        ):
            self.logger.info('Waiting for the plugin to stop')
            try:
                if self._thread:
                    self._thread.join(timeout=self._stop_timeout)
                    if self._thread and self._thread.is_alive():
                        self.logger.warning(
                            'Timeout (seconds=%s) on exit for the plugin',
                            self._stop_timeout,
                        )
            except Exception as e:
                self.logger.warning('Could not join thread on stop: %s', e)

        self.logger.info(
            'Stopped plugin: [%s]', get_plugin_name_by_class(self.__class__)
        )

    def _runner(self):
        """
        Implementation of the runner thread.
        """
        if self.disable_monitor:
            return

        self.logger.info(
            'Starting plugin: [%s]', get_plugin_name_by_class(self.__class__)
        )

        while not self.should_stop():
            try:
                self.main()
            except Exception as e:
                self.logger.exception(e)

            if self.poll_interval:
                self.wait_stop(self.poll_interval)

        self._thread = None


class AsyncRunnablePlugin(RunnablePlugin, ABC):
    """
    Class for runnable plugins with an asynchronous event loop attached.
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._loop: Optional[asyncio.AbstractEventLoop] = asyncio.new_event_loop()
        self._task: Optional[asyncio.Task] = None

    @property
    def _should_start_runner(self):
        """
        This property is used to determine if the runner and the event loop
        should be started for this plugin.
        """
        return True

    @abstractmethod
    async def listen(self):
        """
        Main body of the async plugin. When it's called, the event loop should
        already be running and available over `self._loop`.

        :meta private:
        """

    async def _listen(self):
        """
        Wrapper for :meth:`.listen` that catches any exceptions and logs them.
        """
        try:
            await self.listen()
        except KeyboardInterrupt:
            pass
        except RuntimeError as e:
            if not (
                str(e).startswith('Event loop stopped before ')
                or str(e).startswith('no running event loop')
            ):
                raise e

    def _run_listener(self):
        """
        Initialize an event loop and run the listener as a task.
        """
        if not (self._loop):
            raise AssertionError('The loop is not initialized')
        asyncio.set_event_loop(self._loop)

        self._task = self._loop.create_task(self._listen())
        if hasattr(self._task, 'set_name'):
            self._task.set_name(self.__class__.__name__ + '.listen')
        try:
            self._loop.run_until_complete(self._task)
        except Exception as e:
            if not self.should_stop():
                self.logger.warning('The loop has terminated with an error')
                self.logger.exception(e)

        self._task.cancel()

    def main(self):
        if self.should_stop():
            self.logger.info('The plugin is already scheduled to stop')
            return

        self._loop = asyncio.new_event_loop()

        if self._should_start_runner:
            while not self.should_stop():
                try:
                    self._run_listener()
                finally:
                    self.wait_stop(self.poll_interval)
        else:
            self.wait_stop()

    def stop(self):
        if self._loop and self._loop.is_running():
            self._loop.call_soon_threadsafe(self._loop.stop)
            self._loop = None

        super().stop()


class PluginRegistry:
    """
    Registry for plugins, used to keep track of the loaded plugins and their
    configurations.
    """

    _obj: Optional['PluginRegistry'] = None
    """ Singleton instance of the plugin registry. """

    _singleton_lock = RLock()
    """ Lock for the singleton instance creation. """

    def __init__(self):
        self.plugins: Dict[str, Plugin] = {}
        self._lock = RLock()

    def register(self, plugin: Plugin):
        """
        Register a plugin in the registry.

        :param plugin: The plugin to register.
        """
        with self._lock:
            self.plugins[get_plugin_name_by_class(plugin.__class__)] = plugin

    def unregister(self, plugin: Union[Plugin, str]):
        """
        Unregister a plugin from the registry.

        :param plugin: The plugin to unregister.
        """
        with self._lock:
            if isinstance(plugin, Plugin):
                plugin_name = get_plugin_name_by_class(plugin.__class__)
            else:
                plugin_name = plugin

            if plugin_name in self.plugins:
                del self.plugins[plugin_name]
            else:
                _logger.warning('Plugin %s not found in the registry', plugin_name)

    @classmethod
    def get(cls) -> 'PluginRegistry':
        """
        Get the singleton instance of the plugin registry.

        :return: The plugin registry instance.
        """
        with cls._singleton_lock:
            if cls._obj is None:
                cls._obj = PluginRegistry()
            return cls._obj

    @classmethod
    def get_plugins_by_type(cls, plugin_type: type[Plugin]) -> list[Plugin]:
        """
        Get all the plugins of a specific type.

        :param plugin_type: The type of the plugins to get.
        :return: A list of plugins of the specified type.
        """
        return [
            plugin
            for plugin in cls.get().plugins.values()
            if isinstance(plugin, plugin_type)
        ]


# vim:sw=4:ts=4:et:
