iLSP/ilsp/bicep_lsp/modules.py

"""
LRU Bicep module catalog — loaded from the bundled catalog file at startup.

The catalog (bicep_modules_catalog.json) is baked into the Docker image at build time.
No runtime dependency on DevOpsMCP or any external service.

Provides completion items for:
- LRU module names (when typing a module reference string)
- Module versions per module (when cursor is after 'br/modules:NAME:')
- Module params per version (when cursor is inside a params {} block)
"""

import json
import logging
import pathlib
from typing import Any

logger = logging.getLogger(__name__)

# Catalog is baked into the image root at /bicep_modules_catalog.json
_CATALOG_PATHS = [
    pathlib.Path("/data/bicep_modules_catalog.json"),       # volume-mount (freshest)
    pathlib.Path("/bicep_modules_catalog.json"),             # baked into image (fallback)
    pathlib.Path(__file__).parent.parent.parent / "bicep_modules_catalog.json",  # dev
]


def _load_catalog() -> list[dict[str, Any]]:
    """Load modules from the bundled catalog file, preserving per-version schema."""
    for path in _CATALOG_PATHS:
        if path.exists():
            try:
                data = json.loads(path.read_text())
                modules_raw = data.get("modules", {})
                registry = data.get("registry", "iactemplatereg.azurecr.io")
                modules = []
                for mod_path, info in modules_raw.items():
                    versions = info.get("versions", ["latest"])
                    name = mod_path.split("/")[-1] if "/" in mod_path else mod_path
                    modules.append({
                        "name": name,
                        "path": mod_path,
                        "versions": versions,
                        "latest": versions[-1] if versions else "latest",
                        "registry": registry,
                        "schema": info.get("schema", {}),  # per-version → params
                    })
                logger.info("Bicep catalog loaded from %s: %d modules", path, len(modules))
                return modules
            except Exception:
                logger.exception("Failed to parse catalog at %s", path)
    logger.warning("No bicep_modules_catalog.json found — completions disabled")
    return []


class BicepModuleCatalog:
    """In-memory catalog of LRU Bicep modules, loaded once at startup."""

    _modules: list[dict[str, Any]] = []

    @classmethod
    def load(cls) -> None:
        """Load catalog from disk. Call once at startup."""
        cls._modules = _load_catalog()

    @classmethod
    def get_modules(cls) -> list[dict[str, Any]]:
        return cls._modules

    @classmethod
    def get_module_by_name(cls, name: str) -> dict[str, Any] | None:
        for mod in cls._modules:
            if mod["name"] == name:
                return mod
        return None

    @classmethod
    def as_completion_items(cls) -> list[dict[str, Any]]:
        """Module name completions — shown when typing a module reference string."""
        items = []
        for mod in cls._modules:
            ref = f"br/modules:{mod['path']}:{mod['latest']}"
            items.append({
                "label": mod["name"],
                "kind": 9,  # Module
                "detail": f"LRU Bicep module — {mod['registry']}",
                "insertText": ref,
                "sortText": f"0_lru_{mod['name']}",
                "documentation": {
                    "kind": "markdown",
                    "value": (
                        f"**{mod['name']}** (LRU internal)\n\n"
                        f"Registry: `{mod['registry']}`\n"
                        f"Versions: {', '.join(mod['versions'])}\n\n"
                        f"```bicep\nmodule {mod['name'].lower()} '{ref}' = {{\n"
                        f"  name: '{mod['name'].lower()}'\n  params: {{}}\n}}\n```"
                    ),
                },
            })
        return items

    @classmethod
    def version_completion_items(cls, module_name: str) -> list[dict[str, Any]]:
        """Version completions for a specific module (newest first)."""
        mod = cls.get_module_by_name(module_name)
        if not mod:
            return []
        schema = mod.get("schema", {})
        items = []
        for ver in reversed(mod["versions"]):
            ver_schema = schema.get(ver, {})
            params = list(ver_schema.get("parameters", {}).keys())
            param_summary = (
                f"{len(params)} params: {', '.join(params[:3])}{'...' if len(params) > 3 else ''}"
                if params else "no params"
            )
            items.append({
                "label": ver,
                "kind": 12,  # Value
                "detail": param_summary,
                "insertText": ver,
                "sortText": f"0_lru_ver_{ver}",
                "documentation": {
                    "kind": "markdown",
                    "value": (
                        f"**{module_name} `{ver}`**\n\n"
                        + (
                            "Params: " + ", ".join(f"`{p}`" for p in params)
                            if params else "_No params_"
                        )
                    ),
                },
            })
        return items

    @classmethod
    def param_completion_items(cls, module_name: str, version: str) -> list[dict[str, Any]]:
        """Param completions for a specific module+version combination."""
        mod = cls.get_module_by_name(module_name)
        if not mod:
            return []
        schema = mod.get("schema", {})

        # Exact version match first, then fall back to closest available
        ver_params = schema.get(version, {}).get("parameters", {})
        if not ver_params:
            for v in reversed(list(schema.keys())):
                candidate = schema[v].get("parameters", {})
                if candidate:
                    ver_params = candidate
                    logger.debug("Param fallback: %s %s→%s", module_name, version, v)
                    break

        items = []
        for param_name, param_info in ver_params.items():
            ptype = param_info.get("type", "any")
            description = param_info.get("description", "").strip()
            allowed = param_info.get("allowed", [])
            doc_lines = [f"**{param_name}** (`{ptype}`)"]
            if description:
                doc_lines.append(f"\n{description}")
            if allowed:
                shown = allowed[:5]
                more = f" + {len(allowed) - 5} more" if len(allowed) > 5 else ""
                doc_lines.append(f"\nAllowed: {', '.join(f'`{a}`' for a in shown)}{more}")
            items.append({
                "label": param_name,
                "kind": 5,  # Field
                "detail": ptype,
                "insertText": f"{param_name}: ",
                "sortText": f"0_lru_param_{param_name}",
                "documentation": {
                    "kind": "markdown",
                    "value": "\n".join(doc_lines),
                },
            })
        return items