nx_post_support/server/src/lsp_server.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
"""Implementation of tool support over LSP."""

from __future__ import annotations

import json
import operator
import os
import pathlib
import re
import sys
import threading
import time
from collections import ChainMap
from functools import reduce
from typing import Any, Optional


# **********************************************************
# Update sys.path before importing any bundled libraries.
# **********************************************************
def update_sys_path(path_to_add: str, strategy: str) -> None:
    """Add given path to `sys.path`."""
    if path_to_add not in sys.path and os.path.isdir(path_to_add):
        if strategy == "useBundled":
            sys.path.insert(0, path_to_add)
        elif strategy == "fromEnvironment":
            sys.path.append(path_to_add)


# Ensure that we can import LSP libraries, and other bundled libraries.
update_sys_path(
    os.fspath(pathlib.Path(__file__).parent.parent / "libs"),
    os.getenv("LS_IMPORT_STRATEGY", "useBundled"),
)

# **********************************************************
# Imports needed for the language server goes below this.
# **********************************************************
# pylint: disable=wrong-import-position,import-error
import lsprotocol.types as lsp
from pygls import uris
from pygls.workspace.text_document import TextDocument

import lsp_jsonrpc as jsonrpc
from common.load_data import standard_items
from lsp_tclserver import TclLanguageServer
from tools.completion_items import (
    CompletionContext,
    array_element_completions,
    completion_context,
    ranked_completion_items,
)
from tools.def_navigation import (
    all_def_target_locations,
    def_declarations,
    def_hover_markdown,
    effective_def_locations,
    def_rename_edits,
    def_symbol_at,
)
from tools.def_flow import derived_def_symbol
from tools.def_symbols import parse_def_document
from tools.folding_ranges import build_folding_ranges
from tools.index_cache import IndexCache
from tools.inlay_hint import (
    InlayHintGenerator,
    build_builtin_inlay_signatures,
)
from tools.navigation import (
    DEF_SYMBOL_KINDS,
    SymbolIdentity,
    call_hierarchy_identity,
    call_hierarchy_items,
    document_highlights,
    incoming_call_hierarchy,
    matching_occurrences,
    outgoing_call_hierarchy,
    symbol_at_position,
    workspace_symbols,
)
from tools.semantic_tokens import (
    TOKEN_TYPE_INDEX,
    TOKEN_TYPES,
    TokenModifier,
    _Highlighter,
)
from tools.signature_help import build_signature_help
from tools.tcloo_arguments import method_signature_help
from tclint.lexer import TclSyntaxError
from tools.tcloo_completion import may_contain_classes, parse_completion_source, tcloo_completions
from tools.tcloo_symbols import class_completion_items
from tools.tcloo_navigation import tcloo_definition
from tools.tcl_command_completion import (
    TCL_COMMAND_ITEMS,
    TCL_COMMAND_NAMES,
    DynamicCompletionKind,
    line_prefix_at_position,
    path_completion_items,
    tcl_argument_completion,
)

WORKSPACE_SETTINGS = {}
GLOBAL_SETTINGS = {}
# Extension storage folder of the workspace; without it nothing is persisted.
INDEX_CACHE_PATH: dict[str, str | None] = {}


MAX_WORKERS = 5
LSP_SERVER = TclLanguageServer(name="NX Postprocessor Support", version="0.0.1", max_workers=MAX_WORKERS)

BUILTIN_PROC_NAMES = {item.label for item in standard_items.tcl_keyword_list + standard_items.nx_procs} | set(TCL_COMMAND_NAMES)
BUILTIN_VARIABLE_NAMES = {item.label for item in standard_items.nx_variables}
_TCL_COMMAND_ITEMS_BY_LABEL = {item.label: item for item in TCL_COMMAND_ITEMS}
_TCL_KEYWORD_ITEMS = [_TCL_COMMAND_ITEMS_BY_LABEL.get(item.label, item) for item in standard_items.tcl_keyword_list]
_STATIC_TCL_LABELS = {item.label for item in standard_items.tcl_keyword_list}
STATIC_COMPLETION_ITEMS = tuple(_TCL_KEYWORD_ITEMS + [item for item in TCL_COMMAND_ITEMS if item.label not in _STATIC_TCL_LABELS] + standard_items.nx_procs + standard_items.nx_variables)
STATIC_VARIABLE_ITEMS = {item.label: item for item in standard_items.nx_variables}
BUILTIN_INLAY_SIGNATURES = build_builtin_inlay_signatures(standard_items.json_data.get("MOM_procs", []))
BUILTIN_HOVER_ITEMS = {}
for _hover_item in standard_items.json_data.get("MOM_procs", []) + standard_items.json_data.get("mom_variables", []):
    BUILTIN_HOVER_ITEMS.setdefault(_hover_item.get("label"), _hover_item)

# **********************************************************
# Tool specific code goes below this.
# **********************************************************


# Delete "Linting features" section if your tool is NOT a linter.
# **********************************************************
# Linting features start here
# **********************************************************

#  See `pylint` implementation for a full featured linter extension:
#  Pylint: https://github.com/microsoft/vscode-pylint/blob/main/bundled/tool


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_OPEN)
def did_open(params: lsp.DidOpenTextDocumentParams) -> None:
    """LSP handler for textDocument/didOpen request."""
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    LSP_SERVER.clear_cache_for_uri(document.uri)
    LSP_SERVER.analyze_document_now(document)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_SAVE)
def did_save(params: lsp.DidSaveTextDocumentParams) -> None:
    """LSP handler for textDocument/didSave request."""
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    LSP_SERVER.analyze_document_now(document)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CLOSE)
def did_close(params: lsp.DidCloseTextDocumentParams) -> None:
    """LSP handler for textDocument/didClose request."""
    uri = params.text_document.uri
    LSP_SERVER.remove_file_state(uri)
    _index_tcl_file_from_disk(uri)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DID_CHANGE)
def did_change(params: lsp.DidChangeTextDocumentParams) -> None:
    """LSP handler for textDocument/didChange request"""
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    LSP_SERVER.clear_cache_for_uri(document.uri)
    LSP_SERVER.schedule_document_analysis(document)


FILE_OPERATION_OPTIONS = lsp.FileOperationRegistrationOptions(
    filters=[
        lsp.FileOperationFilter(
            scheme="file",
            pattern=lsp.FileOperationPattern(glob="**/*"),
        )
    ]
)


def _index_tcl_file_from_disk(uri: str) -> None:
    if not uri.startswith("file:"):
        return

    path = pathlib.Path(uris.to_fs_path(uri))
    if path.suffix.lower() != ".tcl" or not path.is_file():
        return

    try:
        document = TextDocument(uri=uri, language_id="tcl")
        LSP_SERVER.update_poco_completion_for_file(
            document,
            cache_tree=False,
            require_file_exists=True,
            from_disk=True,
        )
    except (OSError, UnicodeError) as error:
        log_warning(f"Could not re-index {path}: {error}")


@LSP_SERVER.feature(lsp.WORKSPACE_DID_DELETE_FILES, FILE_OPERATION_OPTIONS)
def did_delete_files(params: lsp.DeleteFilesParams) -> None:
    for deleted_file in params.files:
        LSP_SERVER.remove_file_state(deleted_file.uri)


@LSP_SERVER.feature(lsp.WORKSPACE_DID_RENAME_FILES, FILE_OPERATION_OPTIONS)
def did_rename_files(params: lsp.RenameFilesParams) -> None:
    for renamed_file in params.files:
        old_path = pathlib.Path(uris.to_fs_path(renamed_file.old_uri))
        new_path = pathlib.Path(uris.to_fs_path(renamed_file.new_uri))
        indexed_paths = LSP_SERVER.indexed_paths_under_uri(renamed_file.old_uri)
        new_index_paths: set[pathlib.Path] = set()

        for indexed_path in indexed_paths:
            if LSP_SERVER.paths_equal(indexed_path, old_path):
                new_index_paths.add(new_path)
            else:
                relative_path = os.path.relpath(indexed_path, old_path)
                new_index_paths.add(new_path / relative_path)

        # Also handles renaming a previously unindexed file to a TCL file.
        if new_path.suffix.lower() == ".tcl":
            new_index_paths.add(new_path)

        LSP_SERVER.remove_file_state(renamed_file.old_uri)
        for new_index_path in new_index_paths:
            _index_tcl_file_from_disk(new_index_path.as_uri())


@LSP_SERVER.feature(lsp.WORKSPACE_DID_CHANGE_WATCHED_FILES)
def did_change_watched_files(params: lsp.DidChangeWatchedFilesParams) -> None:
    """Keep indexes for closed Tcl files synchronized with disk changes."""
    for change in params.changes:
        suffix = pathlib.Path(uris.to_fs_path(change.uri)).suffix.lower()
        if suffix == ".psc":
            _refresh_psc_index()
            continue
        if suffix == ".def":
            LSP_SERVER.refresh_def_symbols(_workspace_roots(), report=log_warning)
            continue
        if change.type == lsp.FileChangeType.Deleted:
            LSP_SERVER.remove_file_state(change.uri)
        else:
            _index_tcl_file_from_disk(change.uri)


def _workspace_roots() -> list[pathlib.Path]:
    folders = LSP_SERVER.workspace.folders
    roots = [pathlib.Path(uris.to_fs_path(uri)) for uri in folders]
    if not roots and LSP_SERVER.workspace.root_path:
        roots = [pathlib.Path(LSP_SERVER.workspace.root_path)]
    return roots


def _refresh_psc_index():
    LSP_SERVER.refresh_psc_scripts(_workspace_roots(), report=log_warning)


@LSP_SERVER.feature(
    lsp.TEXT_DOCUMENT_DIAGNOSTIC,
    lsp.DiagnosticOptions(
        identifier="pull-diagnostics",
        inter_file_dependencies=False,
        workspace_diagnostics=False,
    ),
)
def document_diagnostic(params: lsp.DocumentDiagnosticParams):
    """Return diagnostics for the requested document"""
    uri = params.text_document.uri
    doc = LSP_SERVER.workspace.get_text_document(uri)
    diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri)
    was_cached = diagnostic_state is not None and diagnostic_state[0] == doc.version
    if not was_cached:
        LSP_SERVER.compute_diagnostics(doc)
        diagnostic_state = LSP_SERVER.diagnostic_snapshot(uri)

    if diagnostic_state is None:
        return lsp.FullDocumentDiagnosticReport(items=[])

    version, diagnostics = diagnostic_state
    result_id = f"{uri}@{version}"

    if was_cached and result_id == params.previous_result_id:
        return lsp.UnchangedDocumentDiagnosticReport(result_id)

    return lsp.FullDocumentDiagnosticReport(items=diagnostics, result_id=result_id)


@LSP_SERVER.feature(
    lsp.TEXT_DOCUMENT_COMPLETION,
    lsp.CompletionOptions(trigger_characters=["$", " ", "-", "(", ","], resolve_provider=True),
)
def on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
    result = _on_completion(params)
    doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    symbol_list = _symbol_list_completion(LSP_SERVER.get_lines(doc), params.position)
    if symbol_list is not None and symbol_list.is_incomplete:
        # Re-request while typing so the full list appears at BLOCK_LIST/ADDR_LIST.
        result.items = [*symbol_list.items, *result.items]
        result.is_incomplete = True
    return result


@LSP_SERVER.feature(lsp.COMPLETION_ITEM_RESOLVE)
def on_completion_resolve(item: lsp.CompletionItem) -> lsp.CompletionItem:
    """Show the .def declaration of a selected block template/address, as on hover."""
    target = item.data.get("def") if isinstance(item.data, dict) else None
    if item.documentation is None and isinstance(target, list) and len(target) == 2:
        markdown = def_hover_markdown(LSP_SERVER.def_documents_snapshot(), tuple(target))
        if markdown:
            item.documentation = lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=markdown)
    return item


# Keywords that expand to all loaded .def names: keyword -> (items, description).
SYMBOL_LIST_KEYWORDS = {
    "BLOCK_LIST": (lambda: LSP_SERVER.block_template_items(), "block templates"),
    "ADDR_LIST": (lambda: LSP_SERVER.address_items(), "addresses"),
}
DEF_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
_WORD_BEFORE_CURSOR_RE = re.compile(r"(?<![$\w])[A-Za-z_]+$")


def _symbol_list_completion(source_lines, position: lsp.Position) -> lsp.CompletionList | None:
    """Complete BLOCK_LIST/ADDR_LIST to all loaded block templates/addresses.

    While the typed word is a prefix of a keyword, an incomplete list with the
    keyword itself is returned. Once the keyword is typed, the complete list of
    quoted names replaces it.
    """
    line_prefix = line_prefix_at_position(source_lines, position)
    match = _WORD_BEFORE_CURSOR_RE.search(line_prefix or "")
    if match is None:
        return None
    word = match.group()
    for keyword, (symbol_items, _) in SYMBOL_LIST_KEYWORDS.items():
        if not word.startswith(keyword):
            continue
        items = []
        ranked = ranked_completion_items(((0, item) for item in symbol_items()), CompletionContext.GENERAL)
        for item in ranked:
            item = _quoted_item(item, source_lines, position, word)
            # Keep every name visible for the typed keyword; text after it narrows.
            item.filter_text = f"{keyword}{item.label}"
            items.append(item)
        return lsp.CompletionList(is_incomplete=False, items=items)

    keywords = [
        lsp.CompletionItem(
            label=keyword,
            kind=lsp.CompletionItemKind.Snippet,
            detail=f"Show all loaded {description}",
            insert_text=keyword,
            sort_text=f"000:{keyword.casefold()}",
            command=lsp.Command(title=f"Show {description}", command="editor.action.triggerSuggest"),
        )
        for keyword, (_, description) in SYMBOL_LIST_KEYWORDS.items()
        if keyword.startswith(word)
    ]
    if keywords:
        return lsp.CompletionList(is_incomplete=True, items=keywords)
    return None


def _on_completion(params: lsp.CompletionParams) -> lsp.CompletionList:
    doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    position = params.position
    source_lines = LSP_SERVER.get_lines(doc)

    def current_tree():
        try:
            return LSP_SERVER.get_tree(doc)
        except TclSyntaxError:
            return None

    oo_items = tcloo_completions(source_lines, position, LSP_SERVER.class_snapshot(doc.path), current_tree)
    if oo_items is not None:
        return lsp.CompletionList(is_incomplete=False, items=oo_items)
    array_items = array_element_completions(
        source_lines,
        position,
        LSP_SERVER.navigation_snapshot().values(),
        str(pathlib.Path(uris.to_fs_path(doc.uri))),
    )
    if array_items is not None:
        return lsp.CompletionList(is_incomplete=False, items=array_items)
    context = completion_context(source_lines, position)

    symbol_list = _symbol_list_completion(source_lines, position)
    if symbol_list is not None and symbol_list.is_incomplete is False:
        return symbol_list

    # Variable completion wins inside command arguments. Otherwise prefer the
    # narrow command grammar when the cursor is at a known subcommand/option.
    argument_completion = None
    if context != CompletionContext.VARIABLE:
        argument_completion = tcl_argument_completion(source_lines, position)
        if argument_completion is not None and argument_completion.dynamic_kind is None:
            items = ranked_completion_items(
                ((0, item) for item in argument_completion.items),
                CompletionContext.GENERAL,
            )
            return lsp.CompletionList(is_incomplete=False, items=items)

        if argument_completion is not None and argument_completion.dynamic_kind == DynamicCompletionKind.PATH:
            dynamic_items: tuple[lsp.CompletionItem, ...] = ()
            if doc.uri.startswith("file:"):
                document_path = pathlib.Path(uris.to_fs_path(doc.uri))
                dynamic_items = path_completion_items(
                    document_path.parent,
                    argument_completion,
                    position,
                )
            path_candidates = [(0, item) for item in argument_completion.items]
            path_candidates.extend((10, item) for item in dynamic_items)
            items = ranked_completion_items(
                path_candidates,
                CompletionContext.GENERAL,
            )
            return lsp.CompletionList(is_incomplete=False, items=items)

    # Space and dash are registered only to open command-aware suggestions.
    # Do not display the broad fallback list when such a trigger has no match.
    if (
        argument_completion is None
        and params.context is not None
        and params.context.trigger_kind == lsp.CompletionTriggerKind.TriggerCharacter
        and params.context.trigger_character in {" ", "-", "(", ","}
    ):
        return lsp.CompletionList(is_incomplete=False, items=[])

    try:
        tree = LSP_SERVER.get_tree(doc)
    except TclSyntaxError:
        tree = parse_completion_source(doc.source)
        if tree is None:
            return lsp.CompletionList(is_incomplete=False, items=[])
    globals_set, procs_locals, proc_ranges = LSP_SERVER.variable_index_for_document(doc, tree)

    local_names: set[str] = set()
    for proc_range in proc_ranges:
        end_line = proc_range.end_line or proc_range.start_line
        if proc_range.start_line <= position.line <= end_line:
            local_names.update(procs_locals.get(proc_range.name, set()))
            break

    candidates: list[tuple[int, lsp.CompletionItem]] = []
    candidates.extend((0, item) for item in class_completion_items(tree))
    for name in sorted(local_names - globals_set):
        candidates.append((
            0,
            lsp.CompletionItem(
                label=name,
                kind=lsp.CompletionItemKind.Variable,
                detail="Local variable",
            ),
        ))

    for name in sorted(globals_set):
        item = STATIC_VARIABLE_ITEMS.get(name)
        if item is None:
            item = lsp.CompletionItem(
                label=name,
                kind=lsp.CompletionItemKind.Variable,
                detail="Workspace variable",
            )
        candidates.append((
            100,
            item,
        ))

    filepath = str(pathlib.Path(uris.to_fs_path(doc.uri)))
    items_by_file = LSP_SERVER.completion_items_by_file_snapshot()
    for item_path in sorted(items_by_file, key=lambda path: path.casefold()):
        priority = 100 if LSP_SERVER.paths_equal(item_path, filepath) else 200
        candidates.extend((priority, item) for item in items_by_file[item_path])

    if argument_completion is not None:
        static_candidates = [(0, item) for item in argument_completion.items]
        if argument_completion.dynamic_kind == DynamicCompletionKind.VARIABLE:
            variable_candidates = list(candidates)
            variable_candidates.extend((300, item) for item in standard_items.nx_variables)
            items = ranked_completion_items(
                variable_candidates,
                CompletionContext.VARIABLE,
            )
            # Command options are valid alongside variables, even though they
            # are keywords and must not pass through the variable-only filter.
            items = [*ranked_completion_items(static_candidates, CompletionContext.GENERAL), *items]
            return lsp.CompletionList(is_incomplete=False, items=items)

        def_symbol_items = {
            DynamicCompletionKind.BLOCK_TEMPLATE: LSP_SERVER.block_template_items,
            DynamicCompletionKind.ADDRESS: LSP_SERVER.address_items,
            DynamicCompletionKind.VALUE: list,
        }.get(argument_completion.dynamic_kind)
        if def_symbol_items is not None:
            # MOM_do_template, MOM_force, ... take a .def name, a fixed value (e.g.
            # Always|Once|Off) or a variable holding one.
            def_symbols = [*static_candidates, *((0, item) for item in def_symbol_items())]
            variable_candidates = list(candidates)
            variable_candidates.extend((300, item) for item in standard_items.nx_variables)
            variables = []
            for item in ranked_completion_items(variable_candidates, CompletionContext.VARIABLE):
                # Without a typed "$" the variable must be substituted.
                item.insert_text = f"${item.label}"
                item.filter_text = item.label
                item.text_edit = None
                variables.append(item)
            quoted = [
                _quoted_item(item, source_lines, position, argument_completion.active_prefix)
                for item in ranked_completion_items(def_symbols, CompletionContext.GENERAL)
            ]
            return lsp.CompletionList(is_incomplete=False, items=[*quoted, *variables])

        if argument_completion.dynamic_kind == DynamicCompletionKind.PROCEDURE:
            procedure_kinds = {
                lsp.CompletionItemKind.Constructor,
                lsp.CompletionItemKind.Function,
                lsp.CompletionItemKind.Method,
            }
            procedure_candidates = [(priority, item) for priority, item in candidates if item.kind in procedure_kinds]
            procedure_candidates.extend((300, item) for item in standard_items.nx_procs)
            items = ranked_completion_items(
                [*static_candidates, *procedure_candidates],
                CompletionContext.GENERAL,
            )
            return lsp.CompletionList(is_incomplete=False, items=items)

        if argument_completion.dynamic_kind == DynamicCompletionKind.NAMESPACE:
            namespace_candidates = list(static_candidates)
            prefix_is_absolute = argument_completion.active_prefix.startswith("::")
            for index in LSP_SERVER.navigation_snapshot().values():
                priority = 100 if LSP_SERVER.paths_equal(index.path, filepath) else 200
                for occurrence in index.occurrences:
                    if occurrence.identity.kind != "namespace":
                        continue
                    name = occurrence.identity.name
                    label = name if prefix_is_absolute else name.removeprefix("::")
                    namespace_candidates.append((
                        priority,
                        lsp.CompletionItem(
                            label=label,
                            kind=lsp.CompletionItemKind.Module,
                            detail="Tcl namespace",
                        ),
                    ))
            items = ranked_completion_items(
                namespace_candidates,
                CompletionContext.GENERAL,
            )
            return lsp.CompletionList(is_incomplete=False, items=items)

    candidates.extend((300, item) for item in STATIC_COMPLETION_ITEMS)
    items = ranked_completion_items(candidates, context)
    return lsp.CompletionList(is_incomplete=False, items=items)


def _quoted_item(
    item: lsp.CompletionItem,
    source_lines,
    position: lsp.Position,
    active_prefix: str,
) -> lsp.CompletionItem:
    """Insert ``item`` as a quoted word, replacing quotes the user already typed."""
    line = source_lines[position.line] if position.line < len(source_lines) else ""
    utf16 = line.encode("utf-16-le")
    start = position.character - len(active_prefix.encode("utf-16-le")) // 2
    end = position.character
    opened = start > 0 and utf16[(start - 1) * 2 : start * 2].decode("utf-16-le") == '"'
    if opened:
        start -= 1
        # Also replace a closing quote the editor inserted automatically.
        if utf16[end * 2 : (end + 1) * 2].decode("utf-16-le") == '"':
            end += 1
    item.text_edit = lsp.TextEdit(
        range=lsp.Range(
            start=lsp.Position(line=position.line, character=start),
            end=lsp.Position(line=position.line, character=end),
        ),
        new_text=f'"{item.label}"',
    )
    item.insert_text = None
    item.filter_text = f'"{item.label}' if opened else item.label
    return item


@LSP_SERVER.feature(
    lsp.TEXT_DOCUMENT_SIGNATURE_HELP,
    lsp.SignatureHelpOptions(
        trigger_characters=[" "],
        retrigger_characters=[" "],
    ),
)
def signature_help(params: lsp.SignatureHelpParams) -> lsp.SignatureHelp | None:
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    try:
        tree = LSP_SERVER.get_tree(document)
    except TclSyntaxError:
        tree = parse_completion_source(document.source)
        if tree is None:
            return None
    method_help = method_signature_help(document.source, params.position, LSP_SERVER.class_snapshot(document.path), tree)
    if method_help is not None:
        return method_help

    custom_signatures, custom_docs = LSP_SERVER.proc_metadata_snapshot(document.path)

    return build_signature_help(
        document.source,
        tree,
        params.position,
        custom_signatures,
        custom_docs,
        standard_items.json_data.get("MOM_procs", []),
    )


# @LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DOCUMENT_SYMBOL)
# def document_symbols(params: lsp.DocumentSymbolParams):
#     doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
#     ast = LSP_SERVER.parser.parse(doc.source)
#     symbols = LSP_SERVER.extract_tcl_symbols(ast)

#     return symbols


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DOCUMENT_SYMBOL)
def document_symbols(params: lsp.DocumentSymbolParams):
    from tools.document_symbols import build_document_symbols

    doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    return build_document_symbols(doc.source)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_INLAY_HINT)
def inlay_hints(params: lsp.InlayHintParams):
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    settings = _get_settings_by_document(document)
    if not settings.get("inlayHint", False):
        return []

    inlay_settings = settings.get("inlayHints", {})
    parameter_names = inlay_settings.get("parameterNames", "all")
    if parameter_names == "none":
        return []

    # Reuse cached AST
    tree = LSP_SERVER.get_tree(document)

    # Built-in NX procedures are the fallback. Workspace procedures replace them,
    # and a declaration in the current file wins over duplicate workspace names.
    custom_signatures = LSP_SERVER.custom_inlay_signatures_snapshot(document.path)
    signatures = ChainMap(custom_signatures, BUILTIN_INLAY_SIGNATURES)

    generator = InlayHintGenerator(
        document.source,
        signatures,
        external_classes=LSP_SERVER.class_snapshot(document.path),
        source_lines=LSP_SERVER.get_lines(document),
        requested_range=params.range,
        parameter_names=parameter_names,
        suppress_when_argument_matches_name=inlay_settings.get("suppressWhenArgumentMatchesName", True),
    )
    return generator.generate(tree)


@LSP_SERVER.feature(
    lsp.TEXT_DOCUMENT_SEMANTIC_TOKENS_FULL,
    lsp.SemanticTokensLegend(
        token_types=TOKEN_TYPES,
        token_modifiers=[m.name for m in TokenModifier],
    ),
)
def semantic_tokens(params: lsp.SemanticTokensParams):
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)

    data = []
    plugins = []
    hl = _Highlighter(plugins, LSP_SERVER.custom_function_names_snapshot())

    # Reuse cached AST
    tree = LSP_SERVER.get_tree(document)
    classes = LSP_SERVER.class_snapshot(document.path)
    if may_contain_classes(document.source, classes):
        hl.highlight_classes(tree, classes)
        hl.highlight_methods(tree, document.source, document.uri, classes)
    tree.accept(hl, recurse=True)

    tokens = hl.tokens()
    for token in tokens:
        data.extend([
            token.line,
            token.offset,
            token.length,
            TOKEN_TYPE_INDEX[token.tok_type],
            reduce(operator.or_, token.tok_modifiers, 0),
        ])
    return lsp.SemanticTokens(data=data)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_FOLDING_RANGE)
def folding_ranges(params: lsp.FoldingRangeParams):
    document = LSP_SERVER.workspace.get_text_document(params.text_document.uri)
    tree = LSP_SERVER.get_tree(document)
    return build_folding_ranges(tree)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_HOVER)
def hover(params: lsp.HoverParams) -> lsp.Hover:
    pos = params.position
    document_uri = params.text_document.uri
    document = LSP_SERVER.workspace.get_text_document(document_uri)
    col = params.position.character

    try:
        line = LSP_SERVER.get_lines(document)[pos.line]
    except IndexError:
        return None

    # Do not show hover for proc name in its declaration
    from tools.proc_docs import is_proc_declaration_line

    if is_proc_declaration_line(line, pos.character):
        return None

    # Identify the token under the cursor
    for m in re.finditer(r"\b\w+\b", line):
        if m.start() <= col <= m.end():
            token = m.group(0)
            break
    else:
        return None

    def_hover = _tcl_def_hover(document_uri, pos, token)
    if def_hover is not None:
        return def_hover

    # 1) If token is a known MOM proc/variable, return built-in hover
    match = BUILTIN_HOVER_ITEMS.get(token)
    if match and match.get("kind") == "function":
        label = match.get("label", "")
        parameters = match.get("parameters", [])
        param_lines = "\n".join(f"- `{p['name']}`: {p['desc']}" for p in parameters) or "_None_"
        example_data = match.get("example", [])
        example_md = "\n".join(f"{line}" for line in example_data)
        returns_data = match.get("returns", ["None"])
        returns_md = "\n".join(f"- {line}" for line in returns_data)
        doc_md = f"""\
### 📘 {label}

**Purpose**
{match.get("description", "No description available.")}

**Format**
`{match.get("format", label)}`

**Parameters**
{param_lines}

**Return value**
{returns_md}

**Example**
```tcl
{example_md}"""
        return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=doc_md))

    # 2) Otherwise, check if the token is a custom proc and show its preceding doc block
    # Build a merged map of proc -> docs gathered during initialization and updates
    proc_doc = LSP_SERVER.proc_documentation(token, document.path)
    if proc_doc is not None:
        return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=proc_doc))

    return None


def _def_target(identity: SymbolIdentity) -> tuple[str, str] | None:
    return (identity.kind, identity.name) if identity.kind in DEF_SYMBOL_KINDS else None


def _word_at(uri: str, position: lsp.Position) -> str | None:
    document = LSP_SERVER.workspace.get_text_document(uri)
    try:
        line = LSP_SERVER.get_lines(document)[position.line]
    except IndexError:
        return None
    for match in re.finditer(r"\b\w+\b", line):
        if match.start() <= position.character <= match.end():
            return match.group(0)
    return None


def _tcl_def_symbol(uri: str, position: lsp.Position, token: str | None):
    """The .def target at ``position``: a direct NX command argument or a derived name.

    Returns (documents, target, range) or None.
    """
    documents = LSP_SERVER.def_documents_snapshot()
    # Cheap guard: only names declared in a .def file are analyzed at all.
    kinds = {kind for kind in DEF_SYMBOL_KINDS if token and any(token in document.names(kind) for document in documents.values())}
    if not kinds:
        return None
    context = _navigation_context(uri, position)
    target = context and _def_target(context[3])
    if target:
        return documents, target, context[2].range
    try:
        tree = LSP_SERVER.get_tree(LSP_SERVER.workspace.get_text_document(uri))
    except TclSyntaxError:
        return None
    derived = derived_def_symbol(tree, position, LSP_SERVER.def_wrapper_table())
    if derived is None or derived[1] != token:
        return None
    found, name, range_ = derived
    kind = next((kind for kind in sorted(found & kinds) if def_declarations(documents, (kind, name))), None)
    return (documents, (kind, name), range_) if kind else None


def _tcl_def_hover(uri: str, position: lsp.Position, token: str) -> lsp.Hover | None:
    symbol = _tcl_def_symbol(uri, position, token)
    if symbol is None:
        return None
    documents, target, range_ = symbol
    markdown = def_hover_markdown(documents, target)
    if not markdown:
        return None
    return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=markdown), range=range_)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DEFINITION)
def goto_definition(params: lsp.DefinitionParams):
    """Resolve TclOO declarations, then indexed proc and variable definitions."""
    try:
        workspace = LSP_SERVER.workspace
    except RuntimeError:
        workspace = None
    if workspace is not None:
        document = workspace.get_text_document(params.text_document.uri)
        try:
            tree = LSP_SERVER.get_tree(document)
        except TclSyntaxError:
            tree = None  # tcloo_definition repairs open delimiters itself.
        target = tcloo_definition(document.source, document.uri, params.position,
                                  LSP_SERVER.class_snapshot(document.path), tree)
        if target is not None:
            return [target]
    context = _navigation_context(params.text_document.uri, params.position)
    if context is None:
        symbol = _tcl_def_symbol(params.text_document.uri, params.position, _word_at(params.text_document.uri, params.position))
        return symbol and effective_def_locations(symbol[0], symbol[1]) or None

    indexes, definitions, _, identity = context
    target = _def_target(identity)
    if target is not None:
        return effective_def_locations(LSP_SERVER.def_documents_snapshot(), target) or None
    locations = [lsp.Location(uri=index.uri, range=occurrence.range) for index, occurrence in matching_occurrences(identity, indexes, definitions) if occurrence.is_definition]
    return _sorted_locations(locations) or None


def _navigation_context(uri: str, position: lsp.Position):
    indexes, definitions = LSP_SERVER.navigation_state()
    filepath = str(pathlib.Path(uris.to_fs_path(uri)))
    index = indexes.get(filepath)
    if index is None or LSP_SERVER.index_update_pending(filepath):
        document = LSP_SERVER.workspace.get_text_document(uri)
        LSP_SERVER.update_poco_completion_for_file(document)
        indexes, definitions = LSP_SERVER.navigation_state()
        index = indexes.get(filepath)
    if index is None:
        return None

    result = symbol_at_position(index, position, definitions)
    if result is None:
        return None
    occurrence, identity = result
    return indexes, definitions, occurrence, identity


def _sorted_locations(locations: list[lsp.Location]) -> list[lsp.Location]:
    return sorted(
        locations,
        key=lambda location: (
            location.uri,
            location.range.start.line,
            location.range.start.character,
        ),
    )


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_REFERENCES)
def references(params: lsp.ReferenceParams) -> list[lsp.Location]:
    context = _navigation_context(params.text_document.uri, params.position)
    if context is None:
        symbol = _tcl_def_symbol(params.text_document.uri, params.position, _word_at(params.text_document.uri, params.position))
        if symbol is None:
            return []
        indexes, _ = LSP_SERVER.navigation_state()
        return all_def_target_locations(symbol[0], indexes, symbol[1], params.context.include_declaration)

    indexes, definitions, _, identity = context
    target = _def_target(identity)
    if target is not None:
        return all_def_target_locations(
            LSP_SERVER.def_documents_snapshot(), indexes, target, params.context.include_declaration
        )
    locations = [
        lsp.Location(uri=index.uri, range=occurrence.range)
        for index, occurrence in matching_occurrences(identity, indexes, definitions)
        if params.context.include_declaration or not occurrence.is_definition
    ]
    return _sorted_locations(locations)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_DOCUMENT_HIGHLIGHT)
def document_highlight(params: lsp.DocumentHighlightParams):
    context = _navigation_context(params.text_document.uri, params.position)
    if context is None:
        return []

    indexes, definitions, _, identity = context
    filepath = str(pathlib.Path(uris.to_fs_path(params.text_document.uri)))
    index = indexes.get(filepath)
    if index is None:
        return []
    return document_highlights(index, identity, definitions)


def _is_renamable(
    identity: SymbolIdentity,
    indexes,
    definitions: set[SymbolIdentity],
) -> bool:
    if identity not in definitions or identity.kind not in {"proc", "variable"}:
        return False

    basename = identity.name.rsplit("::", 1)[-1]
    if identity.kind == "proc":
        if basename in BUILTIN_PROC_NAMES:
            return False
        definition_count = sum(occurrence.is_definition and occurrence.identity == identity for index in indexes.values() for occurrence in index.occurrences)
        return definition_count == 1

    return basename not in BUILTIN_VARIABLE_NAMES


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_PREPARE_RENAME)
def prepare_rename(params: lsp.PrepareRenameParams):
    context = _navigation_context(params.text_document.uri, params.position)
    if context is None:
        return None

    indexes, definitions, occurrence, identity = context
    target = _def_target(identity)
    if target is not None:
        if not def_declarations(LSP_SERVER.def_documents_snapshot(), target):
            return None
    elif not _is_renamable(identity, indexes, definitions):
        return None
    return lsp.PrepareRenamePlaceholder(range=occurrence.range, placeholder=occurrence.placeholder)


@LSP_SERVER.feature(
    lsp.TEXT_DOCUMENT_RENAME,
    lsp.RenameOptions(prepare_provider=True),
)
def rename(params: lsp.RenameParams) -> lsp.WorkspaceEdit | None:
    if not DEF_NAME_RE.fullmatch(params.new_name):
        return None

    context = _navigation_context(params.text_document.uri, params.position)
    if context is None:
        return None

    indexes, definitions, _, identity = context
    target = _def_target(identity)
    if target is not None:
        return def_rename_edits(LSP_SERVER.def_documents_snapshot(), indexes, target, params.new_name)
    if not _is_renamable(identity, indexes, definitions):
        return None

    changes: dict[str, list[lsp.TextEdit]] = {}
    seen = set()
    for index, occurrence in matching_occurrences(identity, indexes, definitions):
        key = (
            index.uri,
            occurrence.range.start.line,
            occurrence.range.start.character,
            occurrence.range.end.line,
            occurrence.range.end.character,
        )
        if key in seen:
            continue
        seen.add(key)
        changes.setdefault(index.uri, []).append(lsp.TextEdit(range=occurrence.range, new_text=params.new_name))

    for edits in changes.values():
        edits.sort(
            key=lambda edit: (
                edit.range.start.line,
                edit.range.start.character,
            ),
            reverse=True,
        )
    return lsp.WorkspaceEdit(changes=changes)


# .def files are not synchronized with the server; the client sends their
# current text with each request.
DEF_REQUEST_DEFINITION = "nxPostSupport/def/definition"
DEF_REQUEST_HOVER = "nxPostSupport/def/hover"
DEF_REQUEST_REFERENCES = "nxPostSupport/def/references"
DEF_REQUEST_PREPARE_RENAME = "nxPostSupport/def/prepareRename"
DEF_REQUEST_RENAME = "nxPostSupport/def/rename"


def _def_request_context(params):
    uri = params.textDocument.uri
    position = lsp.Position(line=params.position.line, character=params.position.character)
    path = uris.to_fs_path(uri)
    documents = LSP_SERVER.def_documents_snapshot(path, params.text)
    symbol = def_symbol_at(parse_def_document(params.text), position)
    if symbol is None:
        return None
    target, range_, _ = symbol
    current = next(key for key in documents if LSP_SERVER.paths_equal(key, path))
    return documents, {current: uri}, target, range_


@LSP_SERVER.feature(DEF_REQUEST_DEFINITION)
def def_definition(params) -> list[lsp.Location] | None:
    context = _def_request_context(params)
    if context is None:
        return None
    documents, def_uris, target, _ = context
    return effective_def_locations(documents, target, def_uris) or None


@LSP_SERVER.feature(DEF_REQUEST_HOVER)
def def_hover(params) -> lsp.Hover | None:
    context = _def_request_context(params)
    if context is None:
        return None
    documents, _, target, range_ = context
    markdown = def_hover_markdown(documents, target)
    if markdown is None:
        return None
    return lsp.Hover(lsp.MarkupContent(kind=lsp.MarkupKind.Markdown, value=markdown), range=range_)


@LSP_SERVER.feature(DEF_REQUEST_REFERENCES)
def def_references(params) -> list[lsp.Location]:
    context = _def_request_context(params)
    if context is None:
        return []
    documents, def_uris, target, _ = context
    include_declaration = bool(getattr(params, "includeDeclaration", True))
    indexes, _ = LSP_SERVER.navigation_state()
    return all_def_target_locations(documents, indexes, target, include_declaration, def_uris)


@LSP_SERVER.feature(DEF_REQUEST_PREPARE_RENAME)
def def_prepare_rename(params):
    context = _def_request_context(params)
    if context is None:
        return None
    documents, _, target, range_ = context
    if not def_declarations(documents, target):
        return None
    return lsp.PrepareRenamePlaceholder(range=range_, placeholder=target[1])


@LSP_SERVER.feature(DEF_REQUEST_RENAME)
def def_rename(params) -> lsp.WorkspaceEdit | None:
    if not DEF_NAME_RE.fullmatch(params.newName):
        return None
    context = _def_request_context(params)
    if context is None:
        return None
    documents, def_uris, target, _ = context
    indexes, _ = LSP_SERVER.navigation_state()
    return def_rename_edits(documents, indexes, target, params.newName, def_uris)


@LSP_SERVER.feature(lsp.WORKSPACE_SYMBOL)
def workspace_symbol(params: lsp.WorkspaceSymbolParams):
    return workspace_symbols(LSP_SERVER.navigation_snapshot(), params.query)


@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_PREPARE_CALL_HIERARCHY)
def prepare_call_hierarchy(params: lsp.CallHierarchyPrepareParams):
    context = _navigation_context(params.text_document.uri, params.position)
    if context is None:
        return None

    indexes, _, _, identity = context
    items = call_hierarchy_items(identity, indexes)
    return items or None


@LSP_SERVER.feature(lsp.CALL_HIERARCHY_INCOMING_CALLS)
def incoming_calls(params: lsp.CallHierarchyIncomingCallsParams):
    identity = call_hierarchy_identity(params.item)
    if identity is None:
        return []

    indexes, definitions = LSP_SERVER.navigation_state()
    return incoming_call_hierarchy(identity, indexes, definitions)


@LSP_SERVER.feature(lsp.CALL_HIERARCHY_OUTGOING_CALLS)
def outgoing_calls(params: lsp.CallHierarchyOutgoingCallsParams):
    identity = call_hierarchy_identity(params.item)
    if identity is None:
        return []

    indexes, definitions = LSP_SERVER.navigation_state()
    return outgoing_call_hierarchy(identity, indexes, definitions)


# **********************************************************
# Linting features end here
# **********************************************************


# **********************************************************
# Formatting features start here
# **********************************************************
#  Sample implementations:
#  Black: https://github.com/microsoft/vscode-black-formatter/blob/main/bundled/tool


# **********************************************************
# Formatting features ends here
# **********************************************************
@LSP_SERVER.feature(lsp.TEXT_DOCUMENT_FORMATTING)
def formatting(params: lsp.DocumentFormattingParams) -> list[lsp.TextEdit] | None:
    """LSP handler for textDocument/formatting request."""
    doc = LSP_SERVER.workspace.get_text_document(params.text_document.uri)

    source = doc.source
    start = lsp.Position(line=0, character=0)
    last_line = source.rsplit("\n", 1)[-1]
    end = lsp.Position(line=source.count("\n"), character=len(last_line))
    if GLOBAL_SETTINGS.get("formatter", True):
        source = LSP_SERVER.format(doc, params.options)
    return [
        lsp.TextEdit(
            range=lsp.Range(start=start, end=end),
            new_text=source,
        )
    ]


# **********************************************************
# Required Language Server Initialization and Exit handlers.
# **********************************************************
@LSP_SERVER.feature(lsp.WORKSPACE_DID_CHANGE_CONFIGURATION)
def did_change_configuration(_: lsp.DidChangeConfigurationParams):
    """LSP Handler for Config Changes"""


@LSP_SERVER.feature(lsp.INITIALIZE)
def initialize(params: lsp.InitializeParams) -> lsp.InitializeResult:
    """LSP handler for initialize request."""
    log_to_output(f"CWD Server: {os.getcwd()}")

    paths = "\r\n   ".join(sys.path)
    log_to_output(f"sys.path used to run Server:\r\n   {paths}")

    GLOBAL_SETTINGS.update(**params.initialization_options.get("globalSettings", {}))
    INDEX_CACHE_PATH["path"] = params.initialization_options.get("indexCachePath")

    settings = params.initialization_options["settings"]
    _update_workspace_settings(settings)
    log_to_output(f"Settings used to run Server:\r\n{json.dumps(settings, indent=4, ensure_ascii=False)}\r\n")
    log_to_output(f"Global settings:\r\n{json.dumps(GLOBAL_SETTINGS, indent=4, ensure_ascii=False)}\r\n")
    semantic_tokens_legend = lsp.SemanticTokensLegend(
        token_types=TOKEN_TYPES,
        token_modifiers=[m.name for m in TokenModifier],
    )
    return lsp.InitializeResult(
        capabilities=lsp.ServerCapabilities(
            document_formatting_provider=GLOBAL_SETTINGS.get("formatter", True),
            folding_range_provider=True,
            semantic_tokens_provider=lsp.SemanticTokensOptions(legend=semantic_tokens_legend, full=True, range=False),
            definition_provider=True,
            document_highlight_provider=True,
            references_provider=True,
            rename_provider=lsp.RenameOptions(prepare_provider=True),
            workspace_symbol_provider=True,
            call_hierarchy_provider=True,
        )
    )


@LSP_SERVER.feature(lsp.INITIALIZED)
def initialized(_params: lsp.InitializedParams):
    """Kick off background indexing to avoid blocking initialization."""

    def index_workspace():
        try:
            try:
                root = LSP_SERVER.workspace.root_path
            except RuntimeError:
                root = None
            if not root:
                log_to_output("Background indexing skipped: no workspace folder is open.")
                return
            log_to_output("Background indexing started...")
            started = time.perf_counter()
            LSP_SERVER.index_cache = IndexCache.load(INDEX_CACHE_PATH.get("path"))
            root_path = pathlib.Path(root)
            skipped_directories = {
                ".git",
                ".nox",
                ".venv",
                "dist",
                "node_modules",
                "out",
            }
            tcl_files = (path for path in root_path.rglob("*.tcl") if not any(part.casefold() in skipped_directories for part in path.relative_to(root_path).parts[:-1]))
            for filepath in sorted(tcl_files, key=lambda path: str(path).casefold()):
                try:
                    document = TextDocument(uri=filepath.as_uri(), language_id="tcl")
                    LSP_SERVER.update_poco_completion_for_file(
                        document,
                        cache_tree=False,
                        require_file_exists=True,
                        from_disk=True,
                    )
                except Exception as error:
                    log_to_output(f"Fehler beim Parsen von {filepath}: {error}")
            _refresh_psc_index()
            LSP_SERVER.index_cache.save()
            log_to_output(f"Background indexing completed in {time.perf_counter() - started:.1f}s.")
        except Exception as e:
            log_to_output(f"Background indexing failed: {e}")

    threading.Thread(target=index_workspace, name="nxps-indexer", daemon=True).start()


@LSP_SERVER.feature(lsp.EXIT)
def on_exit(_params: Optional[Any] = None) -> None:
    """Handle clean up on exit."""
    jsonrpc.shutdown_json_rpc()


@LSP_SERVER.feature(lsp.SHUTDOWN)
def on_shutdown(_params: Optional[Any] = None) -> None:
    """Handle clean up on shutdown."""
    jsonrpc.shutdown_json_rpc()


def _get_global_defaults():
    return {
        "path": GLOBAL_SETTINGS.get("path", []),
        "interpreter": GLOBAL_SETTINGS.get("interpreter", [sys.executable]),
        "args": GLOBAL_SETTINGS.get("args", []),
        "importStrategy": GLOBAL_SETTINGS.get("importStrategy", "useBundled"),
        "showNotifications": GLOBAL_SETTINGS.get("showNotifications", "off"),
        "formatter": GLOBAL_SETTINGS.get("formatter", True),
        "inlayHint": GLOBAL_SETTINGS.get("inlayHint", True),
        "inlayHints": GLOBAL_SETTINGS.get(
            "inlayHints",
            {
                "parameterNames": "all",
                "suppressWhenArgumentMatchesName": True,
            },
        ),
    }


def _update_workspace_settings(settings):
    if not settings:
        key = os.getcwd()
        WORKSPACE_SETTINGS[key] = {
            "cwd": key,
            "workspaceFS": key,
            "workspace": uris.from_fs_path(key),
            **_get_global_defaults(),
        }
        return

    for setting in settings:
        key = uris.to_fs_path(setting["workspace"])
        WORKSPACE_SETTINGS[key] = {
            "cwd": key,
            **setting,
            "workspaceFS": key,
        }


def _get_settings_by_path(file_path: pathlib.Path):
    workspaces = {s["workspaceFS"] for s in WORKSPACE_SETTINGS.values()}

    while file_path != file_path.parent:
        str_file_path = str(file_path)
        if str_file_path in workspaces:
            return WORKSPACE_SETTINGS[str_file_path]
        file_path = file_path.parent

    setting_values = list(WORKSPACE_SETTINGS.values())
    return setting_values[0]


def _get_document_key(document: TextDocument):
    if WORKSPACE_SETTINGS:
        document_workspace = pathlib.Path(document.path)
        workspaces = {s["workspaceFS"] for s in WORKSPACE_SETTINGS.values()}

        # Find workspace settings for the given file.
        while document_workspace != document_workspace.parent:
            if str(document_workspace) in workspaces:
                return str(document_workspace)
            document_workspace = document_workspace.parent

    return None


def _get_settings_by_document(document: TextDocument | None):
    if document is None or document.path is None:
        return list(WORKSPACE_SETTINGS.values())[0]

    key = _get_document_key(document)
    if key is None:
        # This is either a non-workspace file or there is no workspace.
        key = os.fspath(pathlib.Path(document.path).parent)
        return {
            "cwd": key,
            "workspaceFS": key,
            "workspace": uris.from_fs_path(key),
            **_get_global_defaults(),
        }

    return WORKSPACE_SETTINGS[str(key)]


# *****************************************************
# Logging and notification.
# *****************************************************
def log_to_output(message: str, msg_type: lsp.MessageType = lsp.MessageType.Log) -> None:
    LSP_SERVER.window_log_message(lsp.LogMessageParams(message=message, type=msg_type))


def log_error(message: str) -> None:
    log_to_output(message, lsp.MessageType.Error)
    if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["onError", "onWarning", "always"]:
        LSP_SERVER.window_show_message(lsp.ShowMessageParams(message=message, type=lsp.MessageType.Error))


def log_warning(message: str) -> None:
    log_to_output(message, lsp.MessageType.Warning)
    if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["onWarning", "always"]:
        LSP_SERVER.window_show_message(lsp.ShowMessageParams(message=message, type=lsp.MessageType.Warning))


def log_always(message: str) -> None:
    log_to_output(message, lsp.MessageType.Info)
    if os.getenv("LS_SHOW_NOTIFICATION", "off") in ["always"]:
        LSP_SERVER.window_show_message(lsp.ShowMessageParams(message=message, type=lsp.MessageType.Info))


# *****************************************************
# Start the server.
# *****************************************************
if __name__ == "__main__":
    LSP_SERVER.start_io()