github-spec-kit/src/specify_cli/integrations/catalog.py

"""Integration catalog — discovery, validation, and upgrade support.

Provides:
- ``IntegrationCatalogEntry`` — single catalog source metadata.
- ``IntegrationCatalog``      — fetches, caches, and searches integration
  catalogs (built-in + community).
- ``IntegrationDescriptor``   — loads and validates ``integration.yml``.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

import yaml
from packaging import version as pkg_version


# ---------------------------------------------------------------------------
# Errors
# ---------------------------------------------------------------------------

class IntegrationCatalogError(Exception):
    """Raised when a catalog operation fails."""


class IntegrationDescriptorError(Exception):
    """Raised when an integration.yml descriptor is invalid."""


# ---------------------------------------------------------------------------
# IntegrationCatalogEntry
# ---------------------------------------------------------------------------

@dataclass
class IntegrationCatalogEntry:
    """Represents a single catalog source in the catalog stack."""

    url: str
    name: str
    priority: int
    install_allowed: bool
    description: str = ""


# ---------------------------------------------------------------------------
# IntegrationCatalog
# ---------------------------------------------------------------------------

class IntegrationCatalog:
    """Manages integration catalog fetching, caching, and searching."""

    DEFAULT_CATALOG_URL = (
        "https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.json"
    )
    COMMUNITY_CATALOG_URL = (
        "https://raw.githubusercontent.com/github/spec-kit/main/integrations/catalog.community.json"
    )
    CACHE_DURATION = 3600  # 1 hour

    def __init__(self, project_root: Path) -> None:
        self.project_root = project_root
        self.cache_dir = project_root / ".specify" / "integrations" / ".cache"

    # -- URL validation ---------------------------------------------------

    @staticmethod
    def _validate_catalog_url(url: str) -> None:
        from urllib.parse import urlparse

        parsed = urlparse(url)
        is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
        if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost):
            raise IntegrationCatalogError(
                f"Catalog URL must use HTTPS (got {parsed.scheme}://). "
                "HTTP is only allowed for localhost."
            )
        if not parsed.netloc:
            raise IntegrationCatalogError(
                "Catalog URL must be a valid URL with a host."
            )

    # -- Catalog stack ----------------------------------------------------

    def _load_catalog_config(
        self, config_path: Path
    ) -> Optional[List[IntegrationCatalogEntry]]:
        """Load catalog stack from a YAML file.

        Returns None when the file does not exist.

        Raises:
            IntegrationCatalogError: on invalid content
        """
        if not config_path.exists():
            return None
        try:
            data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
        except (yaml.YAMLError, OSError, UnicodeError) as exc:
            raise IntegrationCatalogError(
                f"Failed to read catalog config {config_path}: {exc}"
            )
        if not isinstance(data, dict):
            raise IntegrationCatalogError(
                f"Invalid catalog config {config_path}: expected a YAML mapping at the root"
            )
        catalogs_data = data.get("catalogs", [])
        if not isinstance(catalogs_data, list):
            raise IntegrationCatalogError(
                f"Invalid catalog config: 'catalogs' must be a list, "
                f"got {type(catalogs_data).__name__}"
            )
        if not catalogs_data:
            raise IntegrationCatalogError(
                f"Catalog config {config_path} exists but contains no 'catalogs' entries. "
                f"Remove the file to use built-in defaults, or add valid catalog entries."
            )
        entries: List[IntegrationCatalogEntry] = []
        skipped: List[int] = []
        for idx, item in enumerate(catalogs_data):
            if not isinstance(item, dict):
                raise IntegrationCatalogError(
                    f"Invalid catalog entry at index {idx}: "
                    f"expected a mapping, got {type(item).__name__}"
                )
            url = str(item.get("url", "")).strip()
            if not url:
                skipped.append(idx)
                continue
            self._validate_catalog_url(url)
            try:
                priority = int(item.get("priority", idx + 1))
            except (TypeError, ValueError):
                raise IntegrationCatalogError(
                    f"Invalid priority for catalog '{item.get('name', idx + 1)}': "
                    f"expected integer, got {item.get('priority')!r}"
                )
            raw_install = item.get("install_allowed", False)
            if isinstance(raw_install, str):
                install_allowed = raw_install.strip().lower() in ("true", "yes", "1")
            else:
                install_allowed = bool(raw_install)
            entries.append(
                IntegrationCatalogEntry(
                    url=url,
                    name=str(item.get("name", f"catalog-{idx + 1}")),
                    priority=priority,
                    install_allowed=install_allowed,
                    description=str(item.get("description", "")),
                )
            )
        entries.sort(key=lambda e: e.priority)
        if not entries:
            raise IntegrationCatalogError(
                f"Catalog config {config_path} contains {len(catalogs_data)} "
                f"entries but none have valid URLs (entries at indices {skipped} "
                f"were skipped). Each catalog entry must have a 'url' field."
            )
        return entries

    def get_active_catalogs(self) -> List[IntegrationCatalogEntry]:
        """Return the ordered list of active integration catalogs.

        Resolution:
        1. ``SPECKIT_INTEGRATION_CATALOG_URL`` env var
        2. Project ``.specify/integration-catalogs.yml``
        3. User ``~/.specify/integration-catalogs.yml``
        4. Built-in defaults (built-in + community)
        """
        import sys

        env_value = os.environ.get("SPECKIT_INTEGRATION_CATALOG_URL", "").strip()
        if env_value:
            self._validate_catalog_url(env_value)
            if env_value != self.DEFAULT_CATALOG_URL:
                if not getattr(self, "_non_default_catalog_warning_shown", False):
                    print(
                        "Warning: Using non-default integration catalog. "
                        "Only use catalogs from sources you trust.",
                        file=sys.stderr,
                    )
                    self._non_default_catalog_warning_shown = True
            return [
                IntegrationCatalogEntry(
                    url=env_value,
                    name="custom",
                    priority=1,
                    install_allowed=True,
                    description="Custom catalog via SPECKIT_INTEGRATION_CATALOG_URL",
                )
            ]

        project_cfg = self.project_root / ".specify" / "integration-catalogs.yml"
        catalogs = self._load_catalog_config(project_cfg)
        if catalogs is not None:
            return catalogs

        user_cfg = Path.home() / ".specify" / "integration-catalogs.yml"
        catalogs = self._load_catalog_config(user_cfg)
        if catalogs is not None:
            return catalogs

        return [
            IntegrationCatalogEntry(
                url=self.DEFAULT_CATALOG_URL,
                name="default",
                priority=1,
                install_allowed=True,
                description="Built-in catalog of installable integrations",
            ),
            IntegrationCatalogEntry(
                url=self.COMMUNITY_CATALOG_URL,
                name="community",
                priority=2,
                install_allowed=False,
                description="Community-contributed integrations (discovery only)",
            ),
        ]

    # -- Fetching ---------------------------------------------------------

    def _fetch_single_catalog(
        self,
        entry: IntegrationCatalogEntry,
        force_refresh: bool = False,
    ) -> Dict[str, Any]:
        """Fetch one catalog, with per-URL caching."""
        import urllib.error
        import urllib.request

        url_hash = hashlib.sha256(entry.url.encode()).hexdigest()[:16]
        cache_file = self.cache_dir / f"catalog-{url_hash}.json"
        cache_meta = self.cache_dir / f"catalog-{url_hash}-metadata.json"

        if not force_refresh and cache_file.exists() and cache_meta.exists():
            try:
                meta = json.loads(cache_meta.read_text(encoding="utf-8"))
                cached_at = datetime.fromisoformat(meta.get("cached_at", ""))
                if cached_at.tzinfo is None:
                    cached_at = cached_at.replace(tzinfo=timezone.utc)
                age = (datetime.now(timezone.utc) - cached_at).total_seconds()
                if age < self.CACHE_DURATION:
                    return json.loads(cache_file.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, ValueError, KeyError, TypeError, AttributeError, OSError, UnicodeError):
                # Cache is invalid or stale metadata; delete and refetch from source.
                try:
                    cache_file.unlink(missing_ok=True)
                    cache_meta.unlink(missing_ok=True)
                except OSError:
                    pass  # Cache cleanup is best-effort; ignore deletion failures.

        try:
            with urllib.request.urlopen(entry.url, timeout=10) as resp:
                # Validate final URL after redirects
                final_url = resp.geturl()
                if final_url != entry.url:
                    self._validate_catalog_url(final_url)
                catalog_data = json.loads(resp.read())

            if not isinstance(catalog_data, dict):
                raise IntegrationCatalogError(
                    f"Invalid catalog format from {entry.url}: expected a JSON object"
                )
            if (
                "schema_version" not in catalog_data
                or "integrations" not in catalog_data
            ):
                raise IntegrationCatalogError(
                    f"Invalid catalog format from {entry.url}"
                )
            if not isinstance(catalog_data.get("integrations"), dict):
                raise IntegrationCatalogError(
                    f"Invalid catalog format from {entry.url}: 'integrations' must be a JSON object"
                )

            try:
                self.cache_dir.mkdir(parents=True, exist_ok=True)
                cache_file.write_text(json.dumps(catalog_data, indent=2), encoding="utf-8")
                cache_meta.write_text(
                    json.dumps(
                        {
                            "cached_at": datetime.now(timezone.utc).isoformat(),
                            "catalog_url": entry.url,
                        },
                        indent=2,
                    ),
                    encoding="utf-8",
                )
            except OSError:
                pass  # Cache is best-effort; proceed with fetched data
            return catalog_data

        except urllib.error.URLError as exc:
            raise IntegrationCatalogError(
                f"Failed to fetch catalog from {entry.url}: {exc}"
            )
        except json.JSONDecodeError as exc:
            raise IntegrationCatalogError(
                f"Invalid JSON in catalog from {entry.url}: {exc}"
            )

    def _get_merged_integrations(
        self, force_refresh: bool = False
    ) -> List[Dict[str, Any]]:
        """Fetch and merge integrations from all active catalogs.

        Catalogs are processed in the order returned by
        :meth:`get_active_catalogs`.  On conflicts, the first catalog in that
        order wins (lower numeric priority = higher precedence).  Each dict is
        annotated with ``_catalog_name`` and ``_install_allowed``.
        """
        import sys

        active = self.get_active_catalogs()
        merged: Dict[str, Dict[str, Any]] = {}
        any_success = False

        for entry in active:
            try:
                data = self._fetch_single_catalog(entry, force_refresh)
                any_success = True
            except IntegrationCatalogError as exc:
                print(
                    f"Warning: Could not fetch catalog '{entry.name}': {exc}",
                    file=sys.stderr,
                )
                continue

            for integ_id, integ_data in data.get("integrations", {}).items():
                if not isinstance(integ_data, dict):
                    continue
                if integ_id not in merged:
                    merged[integ_id] = {
                        **integ_data,
                        "id": integ_id,
                        "_catalog_name": entry.name,
                        "_install_allowed": entry.install_allowed,
                    }

        if not any_success and active:
            raise IntegrationCatalogError(
                "Failed to fetch any integration catalog"
            )

        return list(merged.values())

    # -- Search / info ----------------------------------------------------

    def search(
        self,
        query: Optional[str] = None,
        tag: Optional[str] = None,
        author: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """Search catalogs for integrations matching the given filters."""
        results: List[Dict[str, Any]] = []
        for item in self._get_merged_integrations():
            author_val = item.get("author", "")
            if not isinstance(author_val, str):
                author_val = str(author_val) if author_val is not None else ""
            if author and author_val.lower() != author.lower():
                continue
            if tag:
                raw_tags = item.get("tags", [])
                tags_list = raw_tags if isinstance(raw_tags, list) else []
                if tag.lower() not in [t.lower() for t in tags_list if isinstance(t, str)]:
                    continue
            if query:
                raw_tags = item.get("tags", [])
                tags_list = raw_tags if isinstance(raw_tags, list) else []
                name_val = item.get("name", "")
                desc_val = item.get("description", "")
                id_val = item.get("id", "")
                haystack = " ".join(
                    [
                        str(name_val) if name_val else "",
                        str(desc_val) if desc_val else "",
                        str(id_val) if id_val else "",
                    ]
                    + [t for t in tags_list if isinstance(t, str)]
                ).lower()
                if query.lower() not in haystack:
                    continue
            results.append(item)
        return results

    def get_integration_info(
        self, integration_id: str
    ) -> Optional[Dict[str, Any]]:
        """Return catalog metadata for a single integration, or None."""
        for item in self._get_merged_integrations():
            if item["id"] == integration_id:
                return item
        return None

    # -- Cache management -------------------------------------------------

    def clear_cache(self) -> None:
        """Remove all cached catalog files."""
        if self.cache_dir.exists():
            for pattern in ("catalog-*.json", "catalog-*-metadata.json"):
                for f in self.cache_dir.glob(pattern):
                    f.unlink(missing_ok=True)


# ---------------------------------------------------------------------------
# IntegrationDescriptor  (integration.yml)
# ---------------------------------------------------------------------------

class IntegrationDescriptor:
    """Loads and validates an ``integration.yml`` descriptor.

    The descriptor mirrors ``extension.yml`` and ``preset.yml``::

        schema_version: "1.0"
        integration:
          id: "my-agent"
          name: "My Agent"
          version: "1.0.0"
          description: "Integration for My Agent"
          author: "my-org"
        requires:
          speckit_version: ">=0.6.0"
          tools: [...]
        provides:
          commands: [...]
          scripts: [...]
    """

    SCHEMA_VERSION = "1.0"
    REQUIRED_TOP_LEVEL = ["schema_version", "integration", "requires", "provides"]

    def __init__(self, descriptor_path: Path) -> None:
        self.path = descriptor_path
        self.data = self._load(descriptor_path)
        self._validate()

    # -- Loading ----------------------------------------------------------

    @staticmethod
    def _load(path: Path) -> dict:
        try:
            with open(path, "r", encoding="utf-8") as fh:
                return yaml.safe_load(fh) or {}
        except yaml.YAMLError as exc:
            raise IntegrationDescriptorError(f"Invalid YAML in {path}: {exc}")
        except FileNotFoundError:
            raise IntegrationDescriptorError(f"Descriptor not found: {path}")
        except (OSError, UnicodeError) as exc:
            raise IntegrationDescriptorError(
                f"Unable to read descriptor {path}: {exc}"
            )

    # -- Validation -------------------------------------------------------

    def _validate(self) -> None:
        if not isinstance(self.data, dict):
            raise IntegrationDescriptorError(
                f"Descriptor root must be a YAML mapping, got {type(self.data).__name__}"
            )
        for field in self.REQUIRED_TOP_LEVEL:
            if field not in self.data:
                raise IntegrationDescriptorError(
                    f"Missing required field: {field}"
                )

        if self.data["schema_version"] != self.SCHEMA_VERSION:
            raise IntegrationDescriptorError(
                f"Unsupported schema version: {self.data['schema_version']} "
                f"(expected {self.SCHEMA_VERSION})"
            )

        integ = self.data["integration"]
        if not isinstance(integ, dict):
            raise IntegrationDescriptorError(
                "'integration' must be a mapping"
            )
        for field in ("id", "name", "version", "description"):
            if field not in integ:
                raise IntegrationDescriptorError(
                    f"Missing integration.{field}"
                )
            if not isinstance(integ[field], str):
                raise IntegrationDescriptorError(
                    f"integration.{field} must be a string, got {type(integ[field]).__name__}"
                )

        if not re.match(r"^[a-z0-9-]+$", integ["id"]):
            raise IntegrationDescriptorError(
                f"Invalid integration ID '{integ['id']}': "
                "must be lowercase alphanumeric with hyphens only"
            )

        try:
            pkg_version.Version(integ["version"])
        except (pkg_version.InvalidVersion, TypeError):
            raise IntegrationDescriptorError(
                f"Invalid version '{integ['version']}'"
            )

        requires = self.data["requires"]
        if not isinstance(requires, dict):
            raise IntegrationDescriptorError(
                "'requires' must be a mapping"
            )
        if "speckit_version" not in requires:
            raise IntegrationDescriptorError(
                "Missing requires.speckit_version"
            )
        if not isinstance(requires["speckit_version"], str) or not requires["speckit_version"].strip():
            raise IntegrationDescriptorError(
                "requires.speckit_version must be a non-empty string"
            )
        tools = requires.get("tools")
        if tools is not None:
            if not isinstance(tools, list):
                raise IntegrationDescriptorError(
                    "requires.tools must be a list"
                )
            for tool in tools:
                if not isinstance(tool, dict):
                    raise IntegrationDescriptorError(
                        "Each requires.tools entry must be a mapping"
                    )
                tool_name = tool.get("name")
                if not isinstance(tool_name, str) or not tool_name.strip():
                    raise IntegrationDescriptorError(
                        "requires.tools entry 'name' must be a non-empty string"
                    )

        provides = self.data["provides"]
        if not isinstance(provides, dict):
            raise IntegrationDescriptorError(
                "'provides' must be a mapping"
            )
        commands = provides.get("commands", [])
        scripts = provides.get("scripts", [])
        if "commands" in provides and not isinstance(commands, list):
            raise IntegrationDescriptorError(
                "Invalid provides.commands: expected a list"
            )
        if "scripts" in provides and not isinstance(scripts, list):
            raise IntegrationDescriptorError(
                "Invalid provides.scripts: expected a list"
            )
        if not commands and not scripts:
            raise IntegrationDescriptorError(
                "Integration must provide at least one command or script"
            )
        for cmd in commands:
            if not isinstance(cmd, dict):
                raise IntegrationDescriptorError(
                    "Each command entry must be a mapping"
                )
            if "name" not in cmd or "file" not in cmd:
                raise IntegrationDescriptorError(
                    "Command entry missing 'name' or 'file'"
                )
            cmd_name = cmd["name"]
            cmd_file = cmd["file"]
            if not isinstance(cmd_name, str) or not cmd_name.strip():
                raise IntegrationDescriptorError(
                    "Command entry 'name' must be a non-empty string"
                )
            if not isinstance(cmd_file, str) or not cmd_file.strip():
                raise IntegrationDescriptorError(
                    "Command entry 'file' must be a non-empty string"
                )
            if os.path.isabs(cmd_file) or ".." in Path(cmd_file).parts or Path(cmd_file).drive or Path(cmd_file).anchor:
                raise IntegrationDescriptorError(
                    f"Command entry 'file' must be a relative path without '..': {cmd_file}"
                )
        for script_entry in scripts:
            if not isinstance(script_entry, str) or not script_entry.strip():
                raise IntegrationDescriptorError(
                    "Script entry must be a non-empty string"
                )
            if os.path.isabs(script_entry) or ".." in Path(script_entry).parts or Path(script_entry).drive or Path(script_entry).anchor:
                raise IntegrationDescriptorError(
                    f"Script entry must be a relative path without '..': {script_entry}"
                )

    # -- Property accessors -----------------------------------------------

    @property
    def id(self) -> str:
        return self.data["integration"]["id"]

    @property
    def name(self) -> str:
        return self.data["integration"]["name"]

    @property
    def version(self) -> str:
        return self.data["integration"]["version"]

    @property
    def description(self) -> str:
        return self.data["integration"]["description"]

    @property
    def requires_speckit_version(self) -> str:
        return self.data["requires"]["speckit_version"]

    @property
    def commands(self) -> List[Dict[str, Any]]:
        return self.data.get("provides", {}).get("commands", [])

    @property
    def scripts(self) -> List[str]:
        return self.data.get("provides", {}).get("scripts", [])

    @property
    def tools(self) -> List[Dict[str, Any]]:
        return self.data.get("requires", {}).get("tools") or []

    def get_hash(self) -> str:
        """SHA-256 hash of the descriptor file."""
        with open(self.path, "rb") as fh:
            return f"sha256:{hashlib.sha256(fh.read()).hexdigest()}"