import asyncio
import collections
import dataclasses
import os
import time
from collections.abc import AsyncIterable
from contextlib import suppress
from typing import (
    Optional,
    List,
    Any,
    TYPE_CHECKING,
    Tuple,
    Literal,
    Set,
    Dict,
    Type,
)
from collections.abc import Mapping, Container, Iterable, Sequence, Callable

import debputy.l10n as l10n
from debputy.commands.debputy_cmd.output import (
    OutputStyle,
    MarkdownOutputStyle,
)
from debputy.dh.dh_assistant import (
    parse_drules_for_addons,
    DhSequencerData,
    extract_dh_addons_from_control,
)
from debputy.filesystem_scan import OSFSROOverlay, VirtualPathBase
from debputy.l10n import Translations
from debputy.linting.lint_util import (
    LintState,
    AsyncLinterImpl,
    AbortTaskError,
    WorkspaceTextEditSupport,
)
from debputy.lsp.apt_cache import AptCache
from debputy.lsp.config.debputy_config import load_debputy_config, DebputyConfig
from debputy.lsp.diagnostics import DiagnosticReport
from debputy.lsp.maint_prefs import (
    MaintainerPreferenceTable,
    determine_effective_preference,
    EffectiveFormattingPreference,
)
from debputy.lsp.text_util import LintCapablePositionCodec
from debputy.lsp.vendoring._deb822_repro import Deb822FileElement, parse_deb822_file
from debputy.packages import (
    SourcePackage,
    BinaryPackage,
    DctrlParser,
)
from debputy.plugin.api.feature_set import PluginProvidedFeatureSet
from debputy.util import _info, _warn, T
from debputy.yaml import MANIFEST_YAML, YAMLError
from debputy.yaml.compat import CommentedMap

if TYPE_CHECKING:
    import lsprotocol.types as types
    from pygls.server import LanguageServer
    from pygls.workspace import TextDocument
    from pygls.uris import from_fs_path, to_fs_path

else:
    import debputy.lsprotocol.types as types

    try:
        from pygls.server import LanguageServer
        from pygls.workspace import TextDocument
        from pygls.uris import from_fs_path, to_fs_path
    except ImportError as e:

        class LanguageServer:
            def __init__(self, *args, **kwargs) -> None:
                """Placeholder to work if pygls is not installed"""
                # Should not be called
                global e
                raise e  # pragma: no cover


@dataclasses.dataclass(slots=True)
class FileCache:
    doc_uri: str
    path: str
    is_open_in_editor: bool | None = None
    last_doc_version: int | None = None
    last_mtime: float | None = None
    is_valid: bool = False

    def _update_cache(self, doc: "TextDocument", source: str) -> None:
        raise NotImplementedError

    def _clear_cache(self) -> None:
        raise NotImplementedError

    def resolve_cache(self, ls: "DebputyLanguageServer") -> bool:
        doc = ls.workspace.text_documents.get(self.doc_uri)
        if doc is None:
            doc = ls.workspace.get_text_document(self.doc_uri)
            is_open = False
        else:
            is_open = True
        new_content: str | None = None
        if is_open:
            last_doc_version = self.last_doc_version
            current_doc_version = doc.version
            if (
                not self.is_open_in_editor
                or last_doc_version is None
                or current_doc_version is None
                or last_doc_version < current_doc_version
            ):
                new_content = doc.source

            self.last_doc_version = doc.version
            self.is_open_in_editor = True
        elif doc.uri.startswith("file://"):
            try:
                with open(self.path) as fd:
                    st = os.fstat(fd.fileno())
                    current_mtime = st.st_mtime
                    last_mtime = self.last_mtime or current_mtime - 1
                    if self.is_open_in_editor or current_mtime > last_mtime:
                        new_content = fd.read()
                    self.last_mtime = current_mtime
            except FileNotFoundError:
                self._clear_cache()
                self.is_valid = False
                return False
        self.is_open_in_editor = is_open
        if new_content is not None:
            self._update_cache(doc, new_content)
        self.is_valid = True
        return True


@dataclasses.dataclass(slots=True)
class Deb822FileCache(FileCache):
    deb822_file: Deb822FileElement | None = None

    def _update_cache(self, doc: "TextDocument", source: str) -> None:
        deb822_file = parse_deb822_file(
            source.splitlines(keepends=True),
            accept_files_with_error_tokens=True,
            accept_files_with_duplicated_fields=True,
        )
        self.deb822_file = deb822_file

    def _clear_cache(self) -> None:
        self.deb822_file = None


@dataclasses.dataclass(slots=True)
class DctrlFileCache(Deb822FileCache):
    dctrl_parser: DctrlParser | None = None
    source_package: SourcePackage | None = None
    binary_packages: Mapping[str, BinaryPackage] | None = None

    def _update_cache(self, doc: "TextDocument", source: str) -> None:
        deb822_file, source_package, binary_packages = (
            self.dctrl_parser.parse_source_debian_control(
                source.splitlines(keepends=True),
                ignore_errors=True,
            )
        )
        self.deb822_file = deb822_file
        self.source_package = source_package
        self.binary_packages = binary_packages

    def _clear_cache(self) -> None:
        super()._clear_cache()
        self.source_package = None
        self.binary_packages = None


@dataclasses.dataclass(slots=True)
class SalsaCICache(FileCache):
    parsed_content: CommentedMap | None = None

    def _update_cache(self, doc: "TextDocument", source: str) -> None:
        try:
            value = MANIFEST_YAML.load(source)
            if isinstance(value, CommentedMap):
                self.parsed_content = value
        except YAMLError:
            pass

    def _clear_cache(self) -> None:
        self.parsed_content = None


@dataclasses.dataclass(slots=True)
class DebianRulesCache(FileCache):
    sequences: set[str] | None = None
    saw_dh: bool = False

    def _update_cache(self, doc: "TextDocument", source: str) -> None:
        sequences = set()
        self.saw_dh = parse_drules_for_addons(
            source.splitlines(),
            sequences,
        )
        self.sequences = sequences

    def _clear_cache(self) -> None:
        self.sequences = None
        self.saw_dh = False


class LSProvidedLintState(LintState):
    def __init__(
        self,
        ls: "DebputyLanguageServer",
        doc: "TextDocument",
        source_root: str,
        debian_dir_path: str,
        dctrl_parser: DctrlParser,
    ) -> None:
        self._ls = ls
        self._doc = doc
        # Cache lines (doc.lines re-splits everytime)
        self._lines = doc.lines
        self._source_root = OSFSROOverlay.create_root_dir(".", source_root)
        debian_dir = self._source_root.get("debian")
        if debian_dir is not None and not debian_dir.is_dir:
            debian_dir = None
        self._debian_dir = debian_dir
        self._diagnostics: list[types.Diagnostic] | None = None
        self._last_emitted_diagnostic_count = 0
        dctrl_file = os.path.join(debian_dir_path, "control")

        if dctrl_file != doc.path:

            self._dctrl_cache: DctrlFileCache = ls.file_cache_for(
                from_fs_path(dctrl_file),
                dctrl_file,
                DctrlFileCache,
                lambda uri, path: DctrlFileCache(uri, path, dctrl_parser=dctrl_parser),
            )
            self._deb822_file: Deb822FileCache = ls.file_cache_for(
                doc.uri,
                doc.path,
                Deb822FileCache,
                lambda uri, path: Deb822FileCache(uri, path),
            )
        else:
            self._dctrl_cache = ls.file_cache_for(
                doc.uri,
                doc.path,
                DctrlFileCache,
                lambda uri, path: DctrlFileCache(uri, path, dctrl_parser=dctrl_parser),
            )
            self._deb822_file = self._dctrl_cache

        self._salsa_ci_caches = [
            ls.file_cache_for(
                from_fs_path(os.path.join(debian_dir_path, p)),
                os.path.join(debian_dir_path, p),
                SalsaCICache,
                lambda uri, path: SalsaCICache(uri, path),
            )
            for p in ("salsa-ci.yml", os.path.join("..", ".gitlab-ci.yml"))
        ]
        drules_path = os.path.join(debian_dir_path, "rules")
        self._drules_cache = ls.file_cache_for(
            from_fs_path(drules_path) if doc.path != drules_path else doc.uri,
            drules_path,
            DebianRulesCache,
            lambda uri, path: DebianRulesCache(uri, path),
        )

    @property
    def plugin_feature_set(self) -> PluginProvidedFeatureSet:
        return self._ls.plugin_feature_set

    @property
    def doc_uri(self) -> str:
        return self._doc.uri

    @property
    def doc_version(self) -> int | None:
        return self._doc.version

    @property
    def source_root(self) -> VirtualPathBase | None:
        return self._source_root

    @property
    def debian_dir(self) -> VirtualPathBase | None:
        return self._debian_dir

    @property
    def path(self) -> str:
        return self._doc.path

    @property
    def content(self) -> str:
        return self._doc.source

    @property
    def lines(self) -> list[str]:
        return self._lines

    @property
    def position_codec(self) -> LintCapablePositionCodec:
        return self._doc.position_codec

    def _resolve_dctrl(self) -> DctrlFileCache | None:
        dctrl_cache = self._dctrl_cache
        dctrl_cache.resolve_cache(self._ls)
        return dctrl_cache

    @property
    def parsed_deb822_file_content(self) -> Deb822FileElement | None:
        cache = self._deb822_file
        cache.resolve_cache(self._ls)
        return cache.deb822_file

    @property
    def source_package(self) -> SourcePackage | None:
        return self._resolve_dctrl().source_package

    @property
    def binary_packages(self) -> Mapping[str, BinaryPackage] | None:
        return self._resolve_dctrl().binary_packages

    def _resolve_salsa_ci(self) -> CommentedMap | None:
        for salsa_ci_cache in self._salsa_ci_caches:
            if salsa_ci_cache.resolve_cache(self._ls):
                return salsa_ci_cache.parsed_content
        return None

    @property
    def effective_preference(self) -> EffectiveFormattingPreference | None:
        source_package = self.source_package
        salsa_ci = self._resolve_salsa_ci()
        if source_package is None and salsa_ci is None:
            return None
        style, _, _ = determine_effective_preference(
            self.maint_preference_table,
            source_package,
            salsa_ci,
        )
        return style

    @property
    def maint_preference_table(self) -> MaintainerPreferenceTable:
        return self._ls.maint_preferences

    @property
    def salsa_ci(self) -> CommentedMap | None:
        return None

    @property
    def dh_sequencer_data(self) -> DhSequencerData:
        dh_sequences: set[str] = set()
        saw_dh = False
        src_pkg = self.source_package
        drules_cache = self._drules_cache
        if drules_cache.resolve_cache(self._ls):
            saw_dh = drules_cache.saw_dh
            if drules_cache.sequences:
                dh_sequences.update(drules_cache.sequences)
        if src_pkg:
            extract_dh_addons_from_control(src_pkg.fields, dh_sequences)

        return DhSequencerData(
            frozenset(dh_sequences),
            saw_dh,
        )

    @property
    def workspace_text_edit_support(self) -> WorkspaceTextEditSupport:
        return self._ls.workspace_text_edit_support

    @property
    def debputy_config(self) -> DebputyConfig:
        return self._ls.debputy_config

    def translation(self, domain: str) -> Translations:
        return self._ls.translation(domain)

    async def slow_iter(
        self,
        iterable: Iterable[T],
        *,
        yield_every: int = 100,
    ) -> AsyncIterable[T]:
        counter = 0
        for value in iterable:
            counter += 1
            if counter >= yield_every:
                await asyncio.sleep(0)
                self._abort_on_outdated_doc_version()
                counter = 0
            yield value
        if counter:
            await asyncio.sleep(0)
            self._abort_on_outdated_doc_version()

    async def run_diagnostics(
        self,
        linter: AsyncLinterImpl,
    ) -> list[types.Diagnostic]:
        if self._diagnostics is not None:
            raise RuntimeError(
                "run_diagnostics cannot be run while it is already running"
            )
        self._diagnostics = diagnostics = []

        await linter(self)

        self._diagnostics = None
        return diagnostics

    def _emit_diagnostic(self, diagnostic: types.Diagnostic) -> None:
        diagnostics = self._diagnostics
        if diagnostics is None:
            raise TypeError("Cannot run emit_diagnostic outside of run_diagnostics")

        diagnostics.append(diagnostic)
        self._last_emitted_diagnostic_count += 1
        if self._last_emitted_diagnostic_count >= 100:
            self._abort_on_outdated_doc_version()
            self._ls.record_diagnostics(
                self.doc_uri,
                self._doc.version,
                diagnostics,
                is_partial=True,
            )
            self._last_emitted_diagnostic_count = 0

    def _abort_on_outdated_doc_version(self) -> None:
        expected_version = self._doc.version
        current_doc = self._ls.workspace.get_text_document(self.doc_uri)
        if current_doc.version != expected_version:
            raise AbortTaskError(
                f"Cancel (obsolete) diagnostics for doc version {expected_version}: document version changed"
            )


def _preference(
    client_preference: list[types.MarkupKind] | None,
    options: Container[types.MarkupKind],
    fallback_kind: types.MarkupKind,
) -> types.MarkupKind:
    if not client_preference:
        return fallback_kind
    for markdown_kind in client_preference:
        if markdown_kind in options:
            return markdown_kind
    return fallback_kind


class DebputyLanguageServer(LanguageServer):

    def __init__(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self._dctrl_parser: DctrlParser | None = None
        self._plugin_feature_set: PluginProvidedFeatureSet | None = None
        self._trust_language_ids: bool | None = None
        self._finished_initialization = False
        self.maint_preferences = MaintainerPreferenceTable({}, {})
        self.apt_cache = AptCache()
        self.background_tasks = set()
        self.client_locale: str | None = None
        self.forced_locale: str | None = None
        self._active_locale: list[str] | None = None
        self._diagnostic_reports: dict[str, DiagnosticReport] = {}
        self.workspace_text_edit_support = WorkspaceTextEditSupport()
        self.debputy_config = load_debputy_config()
        self._file_state_caches: dict[str, dict[type[FileCache], FileCache]] = (
            collections.defaultdict(dict)
        )
        self.hover_output_style = OutputStyle()

    def finish_startup_initialization(self) -> None:
        if self._finished_initialization:
            return

        assert self._dctrl_parser is not None
        assert self._plugin_feature_set is not None
        assert self._trust_language_ids is not None
        self.maint_preferences = self.maint_preferences.load_preferences()
        _info(
            f"Loaded style preferences: {len(self.maint_preferences.maintainer_preferences)} unique maintainer preferences recorded"
        )
        if (
            self.hover_markup_format(
                types.MarkupKind.Markdown, types.MarkupKind.PlainText
            )
            == types.MarkupKind.Markdown
        ):
            self.hover_output_style = MarkdownOutputStyle()
        self._finished_initialization = True

    async def on_initialize(self, params: types.InitializeParams) -> None:
        task = self.loop.create_task(self._load_apt_cache(), name="Index apt cache")
        self.background_tasks.add(task)
        task.add_done_callback(self.background_tasks.discard)
        self.client_locale = params.locale
        if self.forced_locale is not None:
            _info(
                f"Ignoring client locale: {self.client_locale}. Using {self.forced_locale} instead as requested [--force-locale]"
            )
        else:
            _info(
                f"Client locale: {self.client_locale}. Use --force-locale to override"
            )
        _info(f"Cwd: {os.getcwd()}")
        self._update_locale()
        self.workspace_text_edit_support = WorkspaceTextEditSupport(
            supported_resource_operation_edit_kinds=self._supported_resource_operation_edit_kinds,
            supports_document_changes=self._supports_ws_document_changes,
        )
        self._mask_ghost_requests()

    def _mask_ghost_requests(self) -> None:
        try:
            if types.TEXT_DOCUMENT_DOCUMENT_HIGHLIGHT not in self.lsp.fm.features:

                # Work around `kate` bug (https://bugs.kde.org/show_bug.cgi?id=506664) for selected requests
                # that are really annoying (that is, triggers all the time).
                def _ghost_request(*_args: Any) -> None:
                    _info(
                        "Ignoring unsupported request from client that it should not have sent."
                    )

                self.lsp.fm.features[types.TEXT_DOCUMENT_DOCUMENT_HIGHLIGHT] = (
                    _ghost_request
                )
        except (AttributeError, TypeError, KeyError, ValueError) as e:
            _info(
                f"Could install ghost handler, continuing without. Error was: {str(e)}"
            )
        else:
            _info(
                f"Injecting fake {types.TEXT_DOCUMENT_DOCUMENT_HIGHLIGHT} handler to work around bugs like KDE#506664"
            )

    def _update_locale(self) -> None:
        if self.forced_locale is not None:
            self._active_locale = [self.forced_locale]
        elif self.client_locale is not None:
            self._active_locale = [self.client_locale]
        else:
            self._active_locale = None

    def file_cache_for(
        self,
        uri: str,
        path: str,
        cache_type: type[T],
        initializer: Callable[[str, str], T],
    ) -> T:
        inner_cache = self._file_state_caches.get(path)
        result = inner_cache.get(cache_type) if inner_cache else None
        if result is None:
            result = initializer(uri, path)
            if uri not in self.workspace.text_documents:
                # We do not get a proper notification on when to discard the cache.
                # For now, we simply do not cache them at all to avoid "leaking by infinite cache".
                #
                # Note that even if we did cache them, we would have to track whether the files have
                # changed (which we do not), so the cache itself would not be useful.
                return result
            assert isinstance(result, cache_type)
            if inner_cache is None:
                inner_cache = {}
                self._file_state_caches[path] = inner_cache
            inner_cache[cache_type] = result
        else:
            assert isinstance(result, cache_type)
        return result

    def shutdown(self) -> None:
        for task in self.background_tasks:
            _info(f"Cancelling task: {task.get_name()}")
            self.loop.call_soon_threadsafe(task.cancel)
        return super().shutdown()

    def translation(self, domain: str) -> Translations:
        return l10n.translation(
            domain,
            languages=self._active_locale,
        )

    async def _load_apt_cache(self) -> None:
        if self.apt_cache.state in ("loading", "loaded"):
            _info(
                f"The apt cache data is already in state {self.apt_cache.state}, not re-triggering"
            )
            return
        _info("Starting load of apt cache data")
        start = time.time()
        try:
            await self.apt_cache.load()
        except ValueError as ex:
            _warn(f"Could not load apt cache: {ex}")
        else:
            end = time.time()
            _info(
                f"Loading apt cache finished after {end-start} seconds and is now in state {self.apt_cache.state}"
            )

    @property
    def plugin_feature_set(self) -> PluginProvidedFeatureSet:
        res = self._plugin_feature_set
        if res is None:
            raise RuntimeError(
                "Initialization error: The plugin feature set has not been initialized before it was needed."
            )
        return res

    @plugin_feature_set.setter
    def plugin_feature_set(self, plugin_feature_set: PluginProvidedFeatureSet) -> None:
        if self._plugin_feature_set is not None:
            raise RuntimeError(
                "The plugin_feature_set attribute cannot be changed once set"
            )
        self._plugin_feature_set = plugin_feature_set

    @property
    def dctrl_parser(self) -> DctrlParser:
        res = self._dctrl_parser
        if res is None:
            raise RuntimeError(
                "Initialization error: The dctrl_parser has not been initialized before it was needed."
            )
        return res

    @dctrl_parser.setter
    def dctrl_parser(self, parser: DctrlParser) -> None:
        if self._dctrl_parser is not None:
            raise RuntimeError("The dctrl_parser attribute cannot be changed once set")
        self._dctrl_parser = parser

    def lint_state(self, doc: "TextDocument") -> LSProvidedLintState:
        dir_path = os.path.dirname(doc.path)

        while dir_path and dir_path != "/" and os.path.basename(dir_path) != "debian":
            dir_path = os.path.dirname(dir_path)

        source_root = os.path.dirname(dir_path)

        return LSProvidedLintState(self, doc, source_root, dir_path, self.dctrl_parser)

    @property
    def _client_hover_markup_formats(self) -> list[types.MarkupKind] | None:
        try:
            return (
                self.client_capabilities.text_document.hover.content_format
            )  # type: ignore
        except AttributeError:
            return None

    def hover_markup_format(
        self,
        *options: types.MarkupKind,
        fallback_kind: types.MarkupKind = types.MarkupKind.PlainText,
    ) -> types.MarkupKind:
        """Pick the client preferred hover markup format from a set of options

        :param options: The markup kinds possible.
        :param fallback_kind: If no overlapping option was found in the client preferences
          (or client did not announce a value at all), this parameter is returned instead.
        :returns: The client's preferred markup format from the provided options, or,
          (if there is no overlap), the `fallback_kind` value is returned.
        """
        client_preference = self._client_hover_markup_formats
        return _preference(client_preference, frozenset(options), fallback_kind)

    @property
    def _client_completion_item_document_markup_formats(
        self,
    ) -> list[types.MarkupKind] | None:
        try:
            return (
                self.client_capabilities.text_document.completion.completion_item.documentation_format  # type : ignore
            )
        except AttributeError:
            return None

    @property
    def _supports_ws_document_changes(self) -> bool:
        try:
            return (
                self.client_capabilities.workspace.workspace_edit.document_changes  # type : ignore
            )
        except AttributeError:
            return False

    @property
    def _supported_resource_operation_edit_kinds(
        self,
    ) -> Sequence[types.ResourceOperationKind]:
        try:
            return (
                self.client_capabilities.workspace.workspace_edit.resource_operations  # type : ignore
            )
        except AttributeError:
            return []

    def completion_item_document_markup(
        self,
        *options: types.MarkupKind,
        fallback_kind: types.MarkupKind = types.MarkupKind.PlainText,
    ) -> types.MarkupKind:
        """Pick the client preferred completion item documentation markup format from a set of options

        :param options: The markup kinds possible.
        :param fallback_kind: If no overlapping option was found in the client preferences
          (or client did not announce a value at all), this parameter is returned instead.
        :returns: The client's preferred markup format from the provided options, or,
          (if there is no overlap), the `fallback_kind` value is returned.
        """

        client_preference = self._client_completion_item_document_markup_formats
        return _preference(client_preference, frozenset(options), fallback_kind)

    @property
    def trust_language_ids(self) -> bool:
        v = self._trust_language_ids
        if v is None:
            return True
        return v

    @trust_language_ids.setter
    def trust_language_ids(self, new_value: bool) -> None:
        self._trust_language_ids = new_value

    def determine_language_id(
        self,
        doc: "TextDocument",
    ) -> tuple[Literal["editor-provided", "filename"], str, str]:
        lang_id = doc.language_id
        path = doc.path
        try:
            last_idx = path.rindex("debian/")
        except ValueError:
            cleaned_filename = os.path.basename(path)
        else:
            cleaned_filename = path[last_idx:]

        if self.trust_language_ids and lang_id and not lang_id.isspace():
            if lang_id not in ("fundamental",):
                return "editor-provided", lang_id, cleaned_filename
            _info(
                f"Ignoring editor provided language ID: {lang_id} (reverting to filename based detection instead)"
            )

        return "filename", cleaned_filename, cleaned_filename

    def close_document(self, uri: str) -> None:
        path = to_fs_path(uri)
        with suppress(KeyError):
            del self._diagnostic_reports[uri]
        with suppress(KeyError):
            del self._file_state_caches[path]

    async def slow_iter(
        self,
        iterable: Iterable[T],
        *,
        yield_every: int = 100,
    ) -> AsyncIterable[T]:
        counter = 0
        for value in iterable:
            counter += 1
            if counter >= yield_every:
                await asyncio.sleep(0)
                counter = 0
            yield value
        if counter:
            await asyncio.sleep(0)

    def record_diagnostics(
        self,
        doc_uri: str,
        version: int,
        diagnostics: list[types.Diagnostic],
        is_partial: bool,
    ) -> None:
        self._diagnostic_reports[doc_uri] = DiagnosticReport(
            doc_uri,
            version,
            f"{version}@{doc_uri}",
            is_partial,
            diagnostics,
        )
        self.publish_diagnostics(doc_uri, diagnostics, version)

    def diagnostics_in_range(
        self,
        uri: str,
        text_range: types.Range,
    ) -> list[types.Diagnostic] | None:
        report = self._diagnostic_reports.get(uri)
        if report is None or report.is_in_progress:
            return None
        doc = self.workspace.get_text_document(uri)
        if doc.version is not None and doc.version != report.doc_version:
            return None

        return report.diagnostics_in_range(text_range)