import logging
import os
import shlex
import tempfile
import time

from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING

from ..async_executor import CriticalTask
from ..dcs import Leader, Member, RemoteMember
from ..psycopg import quote_ident, quote_literal
from ..utils import deep_compare, unquote

if TYPE_CHECKING:  # pragma: no cover
    from . import Postgresql

logger = logging.getLogger(__name__)


class Bootstrap(object):

    def __init__(self, postgresql: 'Postgresql') -> None:
        self._postgresql = postgresql
        self._running_custom_bootstrap = False

    @property
    def running_custom_bootstrap(self) -> bool:
        return self._running_custom_bootstrap

    @property
    def keep_existing_recovery_conf(self) -> bool:
        return self._running_custom_bootstrap and self._keep_existing_recovery_conf

    @staticmethod
    def process_user_options(tool: str,
                             options: Union[Any, Dict[str, str], List[Union[str, Dict[str, Any]]]],
                             not_allowed_options: Tuple[str, ...],
                             error_handler: Callable[[str], None]) -> List[str]:
        """Format *options* in a list or dictionary format into command line long form arguments.

        .. note::
            The format of the output of this method is to prepare arguments for use in the ``initdb``
            method of `self._postgres`.

        :Example:

            The *options* can be defined as a dictionary of key, values to be converted into arguments:
            >>> Bootstrap.process_user_options('foo', {'foo': 'bar'}, (), print)
            ['--foo=bar']

            Or as a list of single string arguments
            >>> Bootstrap.process_user_options('foo', ['yes'], (), print)
            ['--yes']

            Or as a list of key, value options
            >>> Bootstrap.process_user_options('foo', [{'foo': 'bar'}], (), print)
            ['--foo=bar']

            Or a combination of single and key, values
            >>> Bootstrap.process_user_options('foo', ['yes', {'foo': 'bar'}], (), print)
            ['--yes', '--foo=bar']

            Options that contain spaces are passed as is to ``subprocess.call``
            >>> Bootstrap.process_user_options('foo', [{'foo': 'bar baz'}], (), print)
            ['--foo=bar baz']

            Options that are quoted will be unquoted, so the quotes aren't interpreted
            literally by the postgres command
            >>> Bootstrap.process_user_options('foo', [{'foo': '"bar baz"'}], (), print)
            ['--foo=bar baz']

        .. note::
            The *error_handler* is called when any of these conditions are met:

            * Key, value dictionaries in the list form contains multiple keys.
            * If a key is listed in *not_allowed_options*.
            * If the options list is not in the required structure.

        :param tool: The name of the tool used in error reports to *error_handler*
        :param options: Options to parse as a list of key, values or single values, or a dictionary
        :param not_allowed_options: List of keys that cannot be used in the list of key, value formatted options
        :param error_handler: A function which will be called when an error condition is encountered
        :returns: List of long form arguments to pass to the named tool
        """
        user_options: List[str] = []

        def option_is_allowed(name: str) -> bool:
            ret = name not in not_allowed_options
            if not ret:
                error_handler('{0} option for {1} is not allowed'.format(name, tool))
            return ret

        if isinstance(options, dict):
            for key, val in options.items():
                if key and val:
                    user_options.append('--{0}={1}'.format(key, unquote(val)))
        elif isinstance(options, list):
            for opt in options:
                if isinstance(opt, str) and option_is_allowed(opt):
                    user_options.append('--{0}'.format(opt))
                elif isinstance(opt, dict):
                    keys = list(opt.keys())
                    if len(keys) != 1 or not isinstance(opt[keys[0]], str) or not option_is_allowed(keys[0]):
                        error_handler('Error when parsing {0} key-value option {1}: only one key-value is allowed'
                                      ' and value should be a string'.format(tool, opt[keys[0]]))
                    user_options.append('--{0}={1}'.format(keys[0], unquote(opt[keys[0]])))
                else:
                    error_handler('Error when parsing {0} option {1}: value should be string value'
                                  ' or a single key-value pair'.format(tool, opt))
        else:
            error_handler('{0} options must be list or dict'.format(tool))
        return user_options

    def _initdb(self, config: Any) -> bool:
        self._postgresql.set_state('initializing new cluster')
        not_allowed_options = ('pgdata', 'nosync', 'pwfile', 'sync-only', 'version')

        def error_handler(e: str) -> None:
            raise Exception(e)

        options = self.process_user_options('initdb', config or [], not_allowed_options, error_handler)
        pwfile = None

        if self._postgresql.config.superuser:
            if 'username' in self._postgresql.config.superuser:
                options.append('--username={0}'.format(self._postgresql.config.superuser['username']))
            if 'password' in self._postgresql.config.superuser:
                (fd, pwfile) = tempfile.mkstemp()
                os.write(fd, self._postgresql.config.superuser['password'].encode('utf-8'))
                os.close(fd)
                options.append('--pwfile={0}'.format(pwfile))

        ret = self._postgresql.initdb(*options)
        if pwfile:
            os.remove(pwfile)
        if ret:
            self._postgresql.configure_server_parameters()
        else:
            self._postgresql.set_state('initdb failed')
        return ret

    def _post_restore(self) -> None:
        self._postgresql.config.restore_configuration_files()
        self._postgresql.configure_server_parameters()

        # make sure there is no trigger file or postgres will be automatically promoted
        trigger_file = self._postgresql.config.triggerfile_good_name
        trigger_file = (self._postgresql.config.get('recovery_conf') or {}).get(trigger_file) or 'promote'
        trigger_file = os.path.abspath(os.path.join(self._postgresql.data_dir, trigger_file))
        if os.path.exists(trigger_file):
            os.unlink(trigger_file)

    def _custom_bootstrap(self, config: Any) -> bool:
        self._postgresql.set_state('running custom bootstrap script')
        params = [] if config.get('no_params') else ['--scope=' + self._postgresql.scope,
                                                     '--datadir=' + self._postgresql.data_dir]
        try:
            logger.info('Running custom bootstrap script: %s', config['command'])
            if self._postgresql.cancellable.call(shlex.split(config['command']) + params) != 0:
                self._postgresql.set_state('custom bootstrap failed')
                return False
        except Exception:
            logger.exception('Exception during custom bootstrap')
            return False
        self._post_restore()

        if 'recovery_conf' in config:
            self._postgresql.config.write_recovery_conf(config['recovery_conf'])
        elif not self.keep_existing_recovery_conf:
            self._postgresql.config.remove_recovery_conf()
        return True

    def call_post_bootstrap(self, config: Dict[str, Any]) -> bool:
        """
        runs a script after initdb or custom bootstrap script is called and waits until completion.
        """
        cmd = config.get('post_bootstrap') or config.get('post_init')
        if cmd:
            r = self._postgresql.config.local_connect_kwargs
            connstring = self._postgresql.config.format_dsn(r, True)
            if 'host' not in r:
                # https://www.postgresql.org/docs/current/static/libpq-pgpass.html
                # A host name of localhost matches both TCP (host name localhost) and Unix domain socket
                # (pghost empty or the default socket directory) connections coming from the local machine.
                r['host'] = 'localhost'  # set it to localhost to write into pgpass

            env = self._postgresql.config.write_pgpass(r)
            env['PGOPTIONS'] = '-c synchronous_commit=local -c statement_timeout=0'

            try:
                ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env)
            except OSError:
                logger.error('post_init script %s failed', cmd)
                return False
            if ret != 0:
                logger.error('post_init script %s returned non-zero code %d', cmd, ret)
                return False
        return True

    def create_replica(self, clone_member: Union[Leader, Member, None]) -> Optional[int]:
        """
            create the replica according to the replica_method
            defined by the user.  this is a list, so we need to
            loop through all methods the user supplies
        """

        self._postgresql.set_state('creating replica')
        self._postgresql.schedule_sanity_checks_after_pause()

        is_remote_member = isinstance(clone_member, RemoteMember)

        # get list of replica methods either from clone member or from
        # the config. If there is no configuration key, or no value is
        # specified, use basebackup
        replica_methods = (clone_member.create_replica_methods if is_remote_member
                           else self._postgresql.create_replica_methods) or ['basebackup']

        if clone_member and clone_member.conn_url:
            r = clone_member.conn_kwargs(self._postgresql.config.replication)
            # add the credentials to connect to the replica origin to pgpass.
            env = self._postgresql.config.write_pgpass(r)
            connstring = self._postgresql.config.format_dsn(r, True)
        else:
            connstring = ''
            env = os.environ.copy()
            # if we don't have any source, leave only replica methods that work without it
            replica_methods = [r for r in replica_methods
                               if self._postgresql.replica_method_can_work_without_replication_connection(r)]

        # go through them in priority order
        ret = 1
        for replica_method in replica_methods:
            if self._postgresql.cancellable.is_cancelled:
                break

            method_config = self._postgresql.replica_method_options(replica_method)

            # if the method is basebackup, then use the built-in
            if replica_method == "basebackup":
                ret = self.basebackup(connstring, env, method_config)
                if ret == 0:
                    logger.info("replica has been created using basebackup")
                    # if basebackup succeeds, exit with success
                    break
            else:
                if not self._postgresql.data_directory_empty():
                    if method_config.get('keep_data', False):
                        logger.info('Leaving data directory uncleaned')
                    else:
                        self._postgresql.remove_data_directory()

                cmd = replica_method
                # user-defined method; check for configuration
                # not required, actually
                if method_config:
                    # look to see if the user has supplied a full command path
                    # if not, use the method name as the command
                    cmd = method_config.pop('command', cmd)

                # add the default parameters
                if not method_config.get('no_params', False):
                    method_config.update({"scope": self._postgresql.scope,
                                          "role": "replica",
                                          "datadir": self._postgresql.data_dir,
                                          "connstring": connstring})
                else:
                    for param in ('no_params', 'no_master', 'no_leader', 'keep_data'):
                        method_config.pop(param, None)
                params = ["--{0}={1}".format(arg, val) for arg, val in method_config.items()]
                try:
                    # call script with the full set of parameters
                    ret = self._postgresql.cancellable.call(shlex.split(cmd) + params, env=env)
                    # if we succeeded, stop
                    if ret == 0:
                        logger.info('replica has been created using %s', replica_method)
                        break
                    else:
                        logger.error('Error creating replica using method %s: %s exited with code=%s',
                                     replica_method, cmd, ret)
                except Exception:
                    logger.exception('Error creating replica using method %s', replica_method)
                    ret = 1

        self._postgresql.set_state('stopped')
        return ret

    def basebackup(self, conn_url: str, env: Dict[str, str], options: Dict[str, Any]) -> Optional[int]:
        # creates a replica data dir using pg_basebackup.
        # this is the default, built-in create_replica_methods
        # tries twice, then returns failure (as 1)
        # uses "stream" as the xlog-method to avoid sync issues
        # supports additional user-supplied options, those are not validated
        maxfailures = 2
        ret = 1
        not_allowed_options = ('pgdata', 'format', 'wal-method', 'xlog-method', 'gzip',
                               'version', 'compress', 'dbname', 'host', 'port', 'username', 'password')
        user_options = self.process_user_options('basebackup', options, not_allowed_options, logger.error)

        for bbfailures in range(0, maxfailures):
            if self._postgresql.cancellable.is_cancelled:
                break
            if not self._postgresql.data_directory_empty():
                self._postgresql.remove_data_directory()
            try:
                ret = self._postgresql.cancellable.call([self._postgresql.pgcommand('pg_basebackup'),
                                                         '--pgdata=' + self._postgresql.data_dir, '-X', 'stream',
                                                         '--dbname=' + conn_url] + user_options, env=env)
                if ret == 0:
                    break
                else:
                    logger.error('Error when fetching backup: pg_basebackup exited with code=%s', ret)

            except Exception as e:
                logger.error('Error when fetching backup with pg_basebackup: %s', e)

            if bbfailures < maxfailures - 1:
                logger.warning('Trying again in 5 seconds')
                time.sleep(5)

        return ret

    def clone(self, clone_member: Union[Leader, Member, None]) -> bool:
        """
             - initialize the replica from an existing member (primary or replica)
             - initialize the replica using the replica creation method that
               works without the replication connection (i.e. restore from on-disk
               base backup)
        """

        ret = self.create_replica(clone_member) == 0
        if ret:
            self._post_restore()
        return ret

    def bootstrap(self, config: Dict[str, Any]) -> bool:
        """ Initialize a new node from scratch and start it. """
        pg_hba = config.get('pg_hba', [])
        method = config.get('method') or 'initdb'
        if method != 'initdb' and method in config and 'command' in config[method]:
            self._keep_existing_recovery_conf = config[method].get('keep_existing_recovery_conf')
            self._running_custom_bootstrap = True
            do_initialize = self._custom_bootstrap
        else:
            method = 'initdb'
            do_initialize = self._initdb
        return do_initialize(config.get(method)) and self._postgresql.config.append_pg_hba(pg_hba) \
            and self._postgresql.config.save_configuration_files() and bool(self._postgresql.start())

    def create_or_update_role(self, name: str, password: Optional[str], options: List[str]) -> None:
        options = list(map(str.upper, options))
        if 'NOLOGIN' not in options and 'LOGIN' not in options:
            options.append('LOGIN')

        if password:
            options.extend(['PASSWORD', quote_literal(password)])

        sql = """DO $$
BEGIN
    SET local synchronous_commit = 'local';
    PERFORM * FROM pg_catalog.pg_authid WHERE rolname = {0};
    IF FOUND THEN
        ALTER ROLE {1} WITH {2};
    ELSE
        CREATE ROLE {1} WITH {2};
    END IF;
END;$$""".format(quote_literal(name), quote_ident(name, self._postgresql.connection()), ' '.join(options))
        self._postgresql.query('SET log_statement TO none')
        self._postgresql.query('SET log_min_duration_statement TO -1')
        self._postgresql.query("SET log_min_error_statement TO 'log'")
        self._postgresql.query("SET pg_stat_statements.track_utility to 'off'")
        try:
            self._postgresql.query(sql)
        finally:
            self._postgresql.query('RESET log_min_error_statement')
            self._postgresql.query('RESET log_min_duration_statement')
            self._postgresql.query('RESET log_statement')
            self._postgresql.query('RESET pg_stat_statements.track_utility')

    def post_bootstrap(self, config: Dict[str, Any], task: CriticalTask) -> Optional[bool]:
        try:
            postgresql = self._postgresql
            superuser = postgresql.config.superuser
            if 'username' in superuser and 'password' in superuser:
                self.create_or_update_role(superuser['username'], superuser['password'], ['SUPERUSER'])

            task.complete(self.call_post_bootstrap(config))
            if task.result:
                replication = postgresql.config.replication
                self.create_or_update_role(replication['username'], replication.get('password'), ['REPLICATION'])

                rewind = postgresql.config.rewind_credentials
                if not deep_compare(rewind, superuser):
                    self.create_or_update_role(rewind['username'], rewind.get('password'), [])
                    for f in ('pg_ls_dir(text, boolean, boolean)', 'pg_stat_file(text, boolean)',
                              'pg_read_binary_file(text)', 'pg_read_binary_file(text, bigint, bigint, boolean)'):
                        sql = """DO $$
BEGIN
    SET local synchronous_commit = 'local';
    GRANT EXECUTE ON function pg_catalog.{0} TO {1};
END;$$""".format(f, quote_ident(rewind['username'], postgresql.connection()))
                        postgresql.query(sql)

                for name, value in (config.get('users') or {}).items():
                    if all(name != a.get('username') for a in (superuser, replication, rewind)):
                        self.create_or_update_role(name, value.get('password'), value.get('options', []))

                # We were doing a custom bootstrap instead of running initdb, therefore we opened trust
                # access from certain addresses to be able to reach cluster and change password
                if self._running_custom_bootstrap:
                    self._running_custom_bootstrap = False
                    # If we don't have custom configuration for pg_hba.conf we need to restore original file
                    if not postgresql.config.get('pg_hba'):
                        if os.path.exists(postgresql.config.pg_hba_conf):
                            os.unlink(postgresql.config.pg_hba_conf)
                        postgresql.config.restore_configuration_files()
                    postgresql.config.write_postgresql_conf()
                    postgresql.config.replace_pg_ident()

                    # at this point there should be no recovery.conf
                    postgresql.config.remove_recovery_conf()

                    if postgresql.config.hba_file:
                        postgresql.restart()
                    else:
                        postgresql.config.replace_pg_hba()
                        if postgresql.pending_restart:
                            postgresql.restart()
                        else:
                            postgresql.reload()
                            time.sleep(1)  # give a time to postgres to "reload" configuration files
                            postgresql.connection().close()  # close connection to reconnect with a new password
                else:  # initdb
                    # We may want create database and extension for citus
                    self._postgresql.citus_handler.bootstrap()
        except Exception:
            logger.exception('post_bootstrap')
            task.complete(False)
        return task.result
