import glob
import logging
import os
import re
import shutil
import textwrap
import threading

import semver

from localstack import config
from localstack.packages import InstallTarget, Package, PackageInstaller
from localstack.packages.java import java_package
from localstack.services.opensearch import versions
from localstack.utils.archives import download_and_extract_with_retry
from localstack.utils.files import chmod_r, load_file, mkdir, rm_rf, save_file
from localstack.utils.java import (
    java_system_properties_proxy,
    java_system_properties_ssl,
    system_properties_to_cli_args,
)
from localstack.utils.run import run
from localstack.utils.ssl import create_ssl_cert, install_predefined_cert_if_available
from localstack.utils.sync import SynchronizedDefaultDict, retry

LOG = logging.getLogger(__name__)

# the version of opensearch which is used by default
OPENSEARCH_DEFAULT_VERSION = "OpenSearch_3.1"

# See https://docs.aws.amazon.com/opensearch-service/latest/developerguide/supported-plugins.html
OPENSEARCH_PLUGIN_LIST = [
    "ingest-attachment",
    "analysis-kuromoji",
]

# the version of elasticsearch that is pre-seeded into the base image (sync with Dockerfile.base)
ELASTICSEARCH_DEFAULT_VERSION = "Elasticsearch_7.10"

# See https://docs.aws.amazon.com/ja_jp/elasticsearch-service/latest/developerguide/aes-supported-plugins.html
ELASTICSEARCH_PLUGIN_LIST = [
    "analysis-icu",
    "ingest-attachment",
    "analysis-kuromoji",
    "mapper-murmur3",
    "mapper-size",
    "analysis-phonetic",
    "analysis-smartcn",
    "analysis-stempel",
    "analysis-ukrainian",
]
# Default ES modules to exclude (save apprx 66MB in the final image)
ELASTICSEARCH_DELETE_MODULES = ["ingest-geoip"]

_OPENSEARCH_INSTALL_LOCKS = SynchronizedDefaultDict(threading.RLock)


class OpensearchPackage(Package):
    def __init__(self, default_version: str = OPENSEARCH_DEFAULT_VERSION):
        super().__init__(name="OpenSearch", default_version=default_version)

    def _get_installer(self, version: str) -> PackageInstaller:
        if version in versions._prefixed_elasticsearch_install_versions:
            if version.startswith("Elasticsearch_5.") or version.startswith("Elasticsearch_6."):
                return ElasticsearchLegacyPackageInstaller(version)
            return ElasticsearchPackageInstaller(version)
        else:
            return OpensearchPackageInstaller(version)

    def get_versions(self) -> list[str]:
        return list(versions.install_versions.keys())


class OpensearchPackageInstaller(PackageInstaller):
    def __init__(self, version: str):
        super().__init__("opensearch", version)

    def _install(self, target: InstallTarget):
        # locally import to avoid having a dependency on ASF when starting the CLI
        from localstack.aws.api.opensearch import EngineType
        from localstack.services.opensearch import versions

        version = self._get_opensearch_install_version()
        install_dir = self._get_install_dir(target)
        with _OPENSEARCH_INSTALL_LOCKS[version]:
            if not os.path.exists(install_dir):
                opensearch_url = versions.get_download_url(version, EngineType.OpenSearch)
                install_dir_parent = os.path.dirname(install_dir)
                mkdir(install_dir_parent)
                # download and extract archive
                tmp_archive = os.path.join(
                    config.dirs.cache, f"localstack.{os.path.basename(opensearch_url)}"
                )
                download_and_extract_with_retry(opensearch_url, tmp_archive, install_dir_parent)
                opensearch_dir = glob.glob(os.path.join(install_dir_parent, "opensearch*"))
                if not opensearch_dir:
                    raise Exception(f"Unable to find OpenSearch folder in {install_dir_parent}")
                shutil.move(opensearch_dir[0], install_dir)

                for dir_name in ("data", "logs", "modules", "plugins", "config/scripts"):
                    dir_path = os.path.join(install_dir, dir_name)
                    mkdir(dir_path)
                    chmod_r(dir_path, 0o777)

                parsed_version = semver.VersionInfo.parse(version)

                # setup security based on the version
                self._setup_security(install_dir, parsed_version)

                # Determine network configuration to use for plugin downloads
                sys_props = {
                    **java_system_properties_proxy(),
                    **java_system_properties_ssl(
                        os.path.join(install_dir, "jdk", "bin", "keytool"),
                        {"JAVA_HOME": os.path.join(install_dir, "jdk")},
                    ),
                }
                java_opts = system_properties_to_cli_args(sys_props)

                keystore_binary = os.path.join(install_dir, "bin", "opensearch-keystore")
                if os.path.exists(keystore_binary):
                    # initialize and create the keystore. Concurrent starts of ES will all try to create it at the same
                    # time, and fail with a race condition. Creating once when installing solves the issue without
                    # the need to lock the starts
                    # Ultimately, each cluster should have its own `config` file and maybe not share the same one
                    output = run(
                        [keystore_binary, "create"],
                        env_vars={"OPENSEARCH_JAVA_OPTS": " ".join(java_opts)},
                    )
                    LOG.debug("Keystore init output: %s", output)

                # install other default plugins for opensearch 1.1+
                # https://forum.opensearch.org/t/ingest-attachment-cannot-be-installed/6494/12
                if parsed_version >= "1.1.0":
                    for plugin in OPENSEARCH_PLUGIN_LIST:
                        plugin_binary = os.path.join(install_dir, "bin", "opensearch-plugin")
                        plugin_dir = os.path.join(install_dir, "plugins", plugin)
                        if not os.path.exists(plugin_dir):
                            LOG.info("Installing OpenSearch plugin %s", plugin)

                            def try_install():
                                output = run(
                                    [plugin_binary, "install", "-b", plugin],
                                    env_vars={"OPENSEARCH_JAVA_OPTS": " ".join(java_opts)},
                                )
                                LOG.debug("Plugin installation output: %s", output)

                            # We're occasionally seeing javax.net.ssl.SSLHandshakeException -> add download retries
                            download_attempts = 3
                            try:
                                retry(try_install, retries=download_attempts - 1, sleep=2)
                            except Exception:
                                LOG.warning(
                                    "Unable to download OpenSearch plugin '%s' after %s attempts",
                                    plugin,
                                    download_attempts,
                                )
                                if not os.environ.get("IGNORE_OS_DOWNLOAD_ERRORS"):
                                    raise

    def _setup_security(self, install_dir: str, parsed_version: semver.VersionInfo):
        """
        Prepares the usage of the SecurityPlugin for the different versions of OpenSearch.
        :param install_dir: root installation directory for OpenSearch which should be configured
        :param parsed_version: parsed semantic version of the OpenSearch installation which should be configured
        """
        # create & copy SSL certs to opensearch config dir
        install_predefined_cert_if_available()
        config_path = os.path.join(install_dir, "config")
        _, cert_file_name, key_file_name = create_ssl_cert()
        shutil.copyfile(cert_file_name, os.path.join(config_path, "cert.crt"))
        shutil.copyfile(key_file_name, os.path.join(config_path, "cert.key"))

        # configure the default roles, roles_mappings, and internal_users
        if parsed_version >= "2.0.0":
            # with version 2 of opensearch and the security plugin, the config moved to the root config folder
            security_config_folder = os.path.join(install_dir, "config", "opensearch-security")
        else:
            security_config_folder = os.path.join(
                install_dir, "plugins", "opensearch-security", "securityconfig"
            )

        # no non-default roles (not even the demo roles) should be set up
        roles_path = os.path.join(security_config_folder, "roles.yml")
        save_file(
            file=roles_path,
            permissions=0o666,
            content=textwrap.dedent(
                """\
                _meta:
                  type: "roles"
                  config_version: 2
                """
            ),
        )

        # create the internal user which allows localstack to manage the running instance
        internal_users_path = os.path.join(security_config_folder, "internal_users.yml")
        save_file(
            file=internal_users_path,
            permissions=0o666,
            content=textwrap.dedent(
                """\
                _meta:
                  type: "internalusers"
                  config_version: 2

                # Define your internal users here
                localstack-internal:
                  hash: "$2y$12$ZvpKLI2nsdGj1ResAmlLne7ki5o45XpBppyg9nXF2RLNfmwjbFY22"
                  reserved: true
                  hidden: true
                  backend_roles: []
                  attributes: {}
                  opendistro_security_roles: []
                  static: false
                """
            ),
        )

        # define the necessary roles mappings for the internal user
        roles_mapping_path = os.path.join(security_config_folder, "roles_mapping.yml")
        save_file(
            file=roles_mapping_path,
            permissions=0o666,
            content=textwrap.dedent(
                """\
                _meta:
                  type: "rolesmapping"
                  config_version: 2

                security_manager:
                  hosts: []
                  users:
                    - localstack-internal
                  reserved: false
                  hidden: false
                  backend_roles: []
                  and_backend_roles: []

                all_access:
                  hosts: []
                  users:
                    - localstack-internal
                  reserved: false
                  hidden: false
                  backend_roles: []
                  and_backend_roles: []
                """
            ),
        )

    def _get_install_marker_path(self, install_dir: str) -> str:
        return os.path.join(install_dir, "bin", "opensearch")

    def _get_opensearch_install_version(self) -> str:
        from localstack.services.opensearch import versions

        if config.SKIP_INFRA_DOWNLOADS:
            self.version = OPENSEARCH_DEFAULT_VERSION

        return versions.get_install_version(self.version)


class ElasticsearchPackageInstaller(PackageInstaller):
    def __init__(self, version: str):
        super().__init__("elasticsearch", version)

    def get_java_env_vars(self) -> dict[str, str]:
        install_dir = self.get_installed_dir()
        return {
            "JAVA_HOME": os.path.join(install_dir, "jdk"),
        }

    def _install(self, target: InstallTarget):
        # locally import to avoid having a dependency on ASF when starting the CLI
        from localstack.aws.api.opensearch import EngineType
        from localstack.services.opensearch import versions

        version = self.get_elasticsearch_install_version()
        install_dir = self._get_install_dir(target)
        installed_executable = os.path.join(install_dir, "bin", "elasticsearch")
        if not os.path.exists(installed_executable):
            es_url = versions.get_download_url(version, EngineType.Elasticsearch)
            install_dir_parent = os.path.dirname(install_dir)
            mkdir(install_dir_parent)
            # download and extract archive
            tmp_archive = os.path.join(config.dirs.cache, f"localstack.{os.path.basename(es_url)}")
            download_and_extract_with_retry(es_url, tmp_archive, install_dir_parent)
            elasticsearch_dir = glob.glob(os.path.join(install_dir_parent, "elasticsearch*"))
            if not elasticsearch_dir:
                raise Exception(f"Unable to find Elasticsearch folder in {install_dir_parent}")
            shutil.move(elasticsearch_dir[0], install_dir)

            for dir_name in ("data", "logs", "modules", "plugins", "config/scripts"):
                dir_path = os.path.join(install_dir, dir_name)
                mkdir(dir_path)
                chmod_r(dir_path, 0o777)

            # Determine network configuration to use for plugin downloads
            sys_props = {
                **java_system_properties_proxy(),
                **java_system_properties_ssl(
                    os.path.join(install_dir, "jdk", "bin", "keytool"),
                    self.get_java_env_vars(),
                ),
            }
            java_opts = system_properties_to_cli_args(sys_props)

            # install default plugins
            for plugin in ELASTICSEARCH_PLUGIN_LIST:
                plugin_binary = os.path.join(install_dir, "bin", "elasticsearch-plugin")
                plugin_dir = os.path.join(install_dir, "plugins", plugin)
                if not os.path.exists(plugin_dir):
                    LOG.info("Installing Elasticsearch plugin %s", plugin)

                    def try_install():
                        output = run(
                            [plugin_binary, "install", "-b", plugin],
                            env_vars={"ES_JAVA_OPTS": " ".join(java_opts)},
                        )
                        LOG.debug("Plugin installation output: %s", output)

                    # We're occasionally seeing javax.net.ssl.SSLHandshakeException -> add download retries
                    download_attempts = 3
                    try:
                        retry(try_install, retries=download_attempts - 1, sleep=2)
                    except Exception:
                        LOG.warning(
                            "Unable to download Elasticsearch plugin '%s' after %s attempts",
                            plugin,
                            download_attempts,
                        )
                        if not os.environ.get("IGNORE_ES_DOWNLOAD_ERRORS"):
                            raise

            keystore_binary = os.path.join(install_dir, "bin", "elasticsearch-keystore")
            if os.path.exists(keystore_binary):
                # initialize and create the keystore. Concurrent starts of ES will all try to create it at the same
                # time, and fail with a race condition. Creating once when installing solves the issue without
                # the need to lock the starts
                # Ultimately, each cluster should have its own `config` file and maybe not share the same one
                output = run(
                    [keystore_binary, "create"],
                    env_vars={"ES_JAVA_OPTS": " ".join(java_opts)},
                )
                LOG.debug("Keystore init output: %s", output)

        # delete some plugins to free up space
        for plugin in ELASTICSEARCH_DELETE_MODULES:
            module_dir = os.path.join(install_dir, "modules", plugin)
            rm_rf(module_dir)

        # disable x-pack-ml plugin (not working on Alpine)
        xpack_dir = os.path.join(install_dir, "modules", "x-pack-ml", "platform")
        rm_rf(xpack_dir)

        # patch JVM options file - replace hardcoded heap size settings
        jvm_options_file = os.path.join(install_dir, "config", "jvm.options")
        if os.path.exists(jvm_options_file):
            jvm_options = load_file(jvm_options_file)
            jvm_options_replaced = re.sub(
                r"(^-Xm[sx][a-zA-Z0-9.]+$)", r"# \1", jvm_options, flags=re.MULTILINE
            )
            if jvm_options != jvm_options_replaced:
                save_file(jvm_options_file, jvm_options_replaced)

    def _get_install_marker_path(self, install_dir: str) -> str:
        return os.path.join(install_dir, "bin", "elasticsearch")

    def get_elasticsearch_install_version(self) -> str:
        from localstack.services.opensearch import versions

        if config.SKIP_INFRA_DOWNLOADS:
            return ELASTICSEARCH_DEFAULT_VERSION

        return versions.get_install_version(self.version)


class ElasticsearchLegacyPackageInstaller(ElasticsearchPackageInstaller):
    """
    Specialised package installer for ElasticSearch 5.x and 6.x

    It installs Java during setup because these releases of ES do not have a bundled JDK.
    This should be removed after these versions are dropped in line with AWS EOL, scheduled for Nov 2026.
    https://docs.aws.amazon.com/opensearch-service/latest/developerguide/what-is.html#choosing-version
    """

    # ES 5.x and 6.x require Java 8
    # See: https://www.elastic.co/guide/en/elasticsearch/reference/6.0/zip-targz.html
    JAVA_VERSION = "8"

    def _prepare_installation(self, target: InstallTarget) -> None:
        java_package.get_installer(self.JAVA_VERSION).install(target)

    def get_java_env_vars(self) -> dict[str, str]:
        return {
            "JAVA_HOME": java_package.get_installer(self.JAVA_VERSION).get_java_home(),
        }


opensearch_package = OpensearchPackage(default_version=OPENSEARCH_DEFAULT_VERSION)
elasticsearch_package = OpensearchPackage(default_version=ELASTICSEARCH_DEFAULT_VERSION)
