"""Track registry — predefined wizard profiles that curate a subset of
source-configurable services.

`bootstrapper/tracks.yml` is the source of truth. This module parses,
validates, and exposes it as typed dataclasses, plus the lookup helpers
the wizard step builder (`bootstrapper/ui/textual/integration.py`) and
the CLI surface (`bootstrapper/start.py`) consume.

Pure data + lookup — no Textual, no Click, no Rich imports inside the
hot path. `format_track_list` does import Rich for human-readable
output but is only called from `--list-tracks` and the no-TTY fallback.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

import yaml
from jsonschema import Draft202012Validator

from services.manifests import load_yaml_strict

__all__ = [
    "Track",
    "TrackRegistry",
    "TracksLoadError",
    "UnknownTrackServiceError",
    "load_tracks",
    "ALWAYS_ON_SERVICE_KEYS",
    "is_in_track",
    "synthesize_track_source_args",
    "normalize_service_key",
    "format_track_list",
]


# The wizard-step ``service_key``s a track-skip predicate must NEVER filter:
# the three services that survive ``ServiceDiscovery.discover()`` and are
# asked in every track. Locked manifests and cloud-provider keys are filtered
# out upstream and never see the predicate.
ALWAYS_ON_SERVICE_KEYS = frozenset({"llm-provider", "prometheus", "grafana"})


# ────────────────────────────────────────────────────────────────────
# Paths
# ────────────────────────────────────────────────────────────────────

_HERE = Path(__file__).resolve().parent
_DEFAULT_TRACKS_PATH = _HERE / "tracks.yml"
_SCHEMA_PATH = _HERE / "schemas" / "tracks.schema.json"


# ────────────────────────────────────────────────────────────────────
# Dataclasses
# ────────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class Track:
    """One row from tracks.yml."""
    key: str                          # slug, e.g. "gen-ai-rag"
    display_name: str
    description: str
    # None == "*" sentinel (the "all" track — never filter).
    # Otherwise: folder keys for services curated by this track,
    # on top of the always-on tier.
    services: frozenset[str] | None


@dataclass(frozen=True)
class TrackRegistry:
    """All tracks + the always-on set, both consumed by the wizard."""
    tracks: tuple[Track, ...]            # canonical display order
    by_key: dict[str, Track]             # quick lookup
    always_on: frozenset[str]            # service keys exempted from skip


# ────────────────────────────────────────────────────────────────────
# Errors
# ────────────────────────────────────────────────────────────────────

class TracksLoadError(Exception):
    """Raised when tracks.yml fails to load or validate."""


class UnknownTrackServiceError(TracksLoadError):
    """Raised when a track lists a service key that doesn't exist as a
    services/<name>/ folder or as a wizard-discovered configurable
    service."""


# ────────────────────────────────────────────────────────────────────
# Family alias map
# ────────────────────────────────────────────────────────────────────

# Some services have runtime_sc keys that diverge from their folder name:
#
# 1. Multi-container families: the wizard's ServiceDiscovery anchors on the
#    head container (e.g. `ray-head`, `spark-master`, `airflow-webserver`),
#    while the folder under services/ is named after the family root
#    (`ray`, `spark`, `airflow`).
#
# 2. Single-container divergences: a couple of older services were declared
#    with a runtime_sc key that already differed from the folder name —
#    services/neo4j/ uses `neo4j-graph-db` and services/open-webui/ uses
#    `open-web-ui` (three hyphens). The wizard hands the predicate the
#    runtime_sc key; tracks.yml uses the folder name.
#
# This map bridges both cases at predicate time. Plain folder keys pass
# through unchanged.
_FAMILY_KEY_ALIASES: dict[str, str] = {
    "ray-head": "ray",
    "celery-worker": "celery",
    "spark-master": "spark",
    "airflow-webserver": "airflow",
    "langfuse-web": "langfuse",
    "llm-graph-builder-frontend": "llm-graph-builder",
    "neo4j-graph-db": "neo4j",
    "open-web-ui": "open-webui",
    # Virtual-manifest services: Click CLI param names use underscores
    # (e.g. ``llm_provider`` from ``--llm-provider-source``), but
    # tracks.yml and always_on use the folder/hyphenated form.
    "llm_provider": "llm-provider",
    "stt_provider": "stt-provider",
    "tts_provider": "tts-provider",
    "doc_processor": "doc-processor",
    "vllm_metal": "vllm-metal",
    "blender_mcp": "blender-mcp",
    "fal": "fal",
}


def normalize_service_key(key: str) -> str:
    """Map a wizard svc.key (possibly a multi-container family head or
    runtime_sc-divergent name) to the folder name used in tracks.yml.
    Plain folder keys pass through."""
    return _FAMILY_KEY_ALIASES.get(key, key)


# ────────────────────────────────────────────────────────────────────
# Loader
# ────────────────────────────────────────────────────────────────────

_validator_singleton: Draft202012Validator | None = None

# Cached registry — populated by load_tracks() on first call when no
# explicit path is given. The schema validator is already cached;
# this avoids re-reading + re-parsing tracks.yml on every helper call
# from the wizard hot path (~6 calls per --track invocation).
_REGISTRY_CACHE: "TrackRegistry | None" = None


def _get_validator() -> Draft202012Validator:
    global _validator_singleton
    if _validator_singleton is None:
        schema = json.loads(_SCHEMA_PATH.read_text(encoding="utf-8"))
        _validator_singleton = Draft202012Validator(schema)
    return _validator_singleton


def _canonical_service_keys() -> frozenset[str]:
    """Every services/<name>/ folder key (the universe a track can pick
    from). Accepts both manifest-bearing dirs and README-only ("doc-only")
    folders like services/doc-processor/ and services/stt-provider/ —
    those still appear in the wizard via source_mapping shims and are
    legitimate track members."""
    services_root = _HERE.parent / "services"
    keys: set[str] = set()
    if not services_root.is_dir():
        return frozenset()
    for child in services_root.iterdir():
        if not child.is_dir():
            continue
        if child.name.startswith(("_", ".")):
            continue
        if not (child / "service.yml").exists() and not (child / "README.md").exists():
            continue
        keys.add(child.name)
    return frozenset(keys)


def load_tracks(path: Path | None = None) -> TrackRegistry:
    """Parse and validate the track registry.

    Path defaults to ``bootstrapper/tracks.yml``. Raises
    ``TracksLoadError`` (and subclasses) on schema / cross-validation
    failure.

    The result is cached in ``_REGISTRY_CACHE`` when called without an
    explicit path (the production hot path). Tests that pass a ``tmp_path``
    always bypass the cache and re-parse — so the cache never affects
    test isolation.
    """
    global _REGISTRY_CACHE
    if path is None and _REGISTRY_CACHE is not None:
        return _REGISTRY_CACHE
    p = Path(path) if path is not None else _DEFAULT_TRACKS_PATH
    try:
        raw = load_yaml_strict(p.read_text(encoding="utf-8"))
    except yaml.YAMLError as e:
        raise TracksLoadError(f"{p}: invalid YAML — {e}") from e

    schema_errors = sorted(
        _get_validator().iter_errors(raw),
        key=lambda e: list(e.absolute_path),
    )
    if schema_errors:
        details = "; ".join(
            f"{'.'.join(str(x) for x in e.absolute_path) or '<root>'}: {e.message}"
            for e in schema_errors
        )
        raise TracksLoadError(f"{p}: schema violation(s): {details}")

    # Build Track objects.
    tracks_list: list[Track] = []
    seen_keys: set[str] = set()
    for entry in raw["tracks"]:
        key = entry["key"]
        if key in seen_keys:
            raise TracksLoadError(f"{p}: duplicate track key '{key}'")
        seen_keys.add(key)
        svc_field = entry["services"]
        services: frozenset[str] | None
        if svc_field == "*":
            services = None
        else:
            services = frozenset(svc_field)
        tracks_list.append(Track(
            key=key,
            display_name=entry["display_name"],
            description=entry["description"],
            services=services,
        ))

    # Cross-validate: every listed service must be a real folder key.
    canonical = _canonical_service_keys()
    unknown: list[tuple[str, str]] = []
    for t in tracks_list:
        if t.services is None:
            continue
        for svc in t.services:
            if svc not in canonical:
                unknown.append((t.key, svc))
    if unknown:
        details = ", ".join(f"{tk}:{sv}" for tk, sv in unknown)
        raise UnknownTrackServiceError(
            f"{p}: services not found under services/<name>/: {details}. "
            f"Valid keys: {sorted(canonical)}"
        )

    tracks_tuple = tuple(tracks_list)
    by_key = {t.key: t for t in tracks_tuple}

    registry = TrackRegistry(
        tracks=tracks_tuple,
        by_key=by_key,
        always_on=ALWAYS_ON_SERVICE_KEYS,
    )
    if path is None:
        _REGISTRY_CACHE = registry
    return registry


# ────────────────────────────────────────────────────────────────────
# Predicate
# ────────────────────────────────────────────────────────────────────

def is_in_track(
    track: Track,
    service_key: str,
    *,
    always_on: frozenset[str],
) -> bool:
    """True iff ``service_key`` (a wizard svc.key) is asked/enabled-by-
    default under ``track``.

    Implements the rule from the spec:
      1. Service is in the always-on tier → always in-track.
      2. Track is the "*" sentinel (services is None) → always in-track.
      3. Service is explicitly listed in track.services after family
         alias normalization → in-track.
      4. Otherwise → out-of-track.
    """
    normalized = normalize_service_key(service_key)
    if normalized in always_on:
        return True
    if track.services is None:
        return True
    return normalized in track.services


def synthesize_track_source_args(
    source_args: dict[str, str | None],
    *,
    track_key: str | None,
    registry: TrackRegistry,
    force_disable: bool,
    consumer_declared: frozenset[str] | set[str] = frozenset(),
) -> set[str]:
    """Apply track membership to Click-style ``*_source`` argument dicts.

    Returns the off-track services that the user explicitly overrode. When
    ``force_disable`` is true, off-track services without an explicit value are
    filled with ``"disabled"``. Wizard mode passes ``force_disable=False`` so
    it can record overrides without making source_args look CLI-configured.

    ``consumer_declared`` (#783): cli-style keys (e.g. ``minio_source``) whose
    SOURCE var the consumer manifest explicitly declares in ``env.values``.
    Declared intent beats the track default: those services are exempt from
    the force-disable fill (and recorded as overrides), so a manifest's
    ``MINIO_SOURCE: container`` survives an out-of-track selection instead of
    being silently reverted to ``disabled`` — an explicit CLI flag still wins
    over both.
    """
    overridden_services: set[str] = set()
    if track_key is None:
        return overridden_services
    track = registry.by_key.get(track_key)
    if track is None or track.services is None:
        return overridden_services
    for cli_key in list(source_args.keys()):
        if cli_key.startswith("cloud_"):
            continue
        svc_key = cli_key.removesuffix("_source").replace("_", "-")
        if is_in_track(track, svc_key, always_on=registry.always_on):
            continue
        explicit_value = source_args.get(cli_key)
        # NORMALIZE before recording. Every consumer looks this set up through
        # `normalize_service_key` — `_make_track_skip` does
        # `_norm(service_key) in overridden`, and the off-track dimming does
        # the same — so an un-normalized key never matches for the two aliases
        # whose CLI stem diverges from the folder name: `open_web_ui_source`
        # yields `open-web-ui` (folder `open-webui`) and `neo4j_graph_db_source`
        # yields `neo4j-graph-db` (folder `neo4j`). For those two, a consumer
        # manifest declaration or an explicit `--<svc>-source` flag was
        # silently force-disabled anyway — the #783 contract, and the one
        # `_make_track_skip`'s own docstring states. `ray`, `spark`, `minio`
        # and the rest were unaffected because their stem already equals the
        # folder name, which is also why the existing tests missed it.
        recorded_key = normalize_service_key(svc_key)
        if explicit_value is not None and explicit_value != "disabled":
            overridden_services.add(recorded_key)
        elif explicit_value is None and cli_key in consumer_declared:
            # Consumer-manifest-declared source: leave untouched so the
            # env.values declaration lands in .env unclobbered; record it as
            # an override for the advisory line.
            overridden_services.add(recorded_key)
        elif explicit_value is None and force_disable:
            source_args[cli_key] = "disabled"
    return overridden_services


# ────────────────────────────────────────────────────────────────────
# Human-readable list (--list-tracks)
# ────────────────────────────────────────────────────────────────────

def remark_off_track_rows(
    track_key: str,
    rows,
    *,
    services_info,
    overridden: frozenset[str] = frozenset(),
):
    """Return ``rows`` with ``off_track`` recomputed for ``track_key``.

    Before #1032 the wizard computed this once while building its steps,
    and only when ``--track`` arrived on the CLI — so an interactive pick
    left every row undimmed and the table advertised a service set the
    launch resolver would not produce. Both paths now share this rule, so
    the preview and the resolver cannot diverge.

    An unknown track, or one naming no service list, excludes nothing:
    with no list to exclude against, that is the honest answer rather than
    a guess. Services the user explicitly re-enabled are never excluded.
    """
    try:
        registry = load_tracks()
        track = registry.by_key.get(track_key)
    except Exception:  # noqa: BLE001
        track = None
    from dataclasses import replace

    if track is None or track.services is None:
        return [replace(row, off_track=False) for row in rows]
    excluded = frozenset(
        svc.display_name for svc in services_info
        if not is_in_track(track, svc.key, always_on=registry.always_on)
        and normalize_service_key(svc.key) not in overridden
    )
    return [replace(row, off_track=row.name in excluded) for row in rows]


def format_track_list(registry: TrackRegistry) -> str:
    """Rich-formatted table for ``--list-tracks`` output and the no-TTY
    stdin prompt fallback. Returns a plain string (Rich renders into a
    string capture buffer)."""
    from io import StringIO
    from rich.console import Console
    from rich.table import Table

    buf = StringIO()
    console = Console(file=buf, force_terminal=False, width=100)
    table = Table(title="Available tracks", show_lines=False)
    table.add_column("Key", style="bold cyan")
    table.add_column("Name")
    table.add_column("Track-specific services")

    for t in registry.tracks:
        if t.services is None:
            svc_cell = "* (every configurable service)"
        else:
            svc_cell = ", ".join(sorted(t.services))
        table.add_row(t.key, t.display_name, svc_cell)
    console.print(table)
    console.print(
        "[dim]Prompted in every track: LLM Engine · Prometheus · Grafana · "
        "OpenAI/Anthropic/OpenRouter cloud keys. Prometheus and Grafana remain "
        "disabled unless enabled.[/dim]"
    )
    return buf.getvalue()


def consumer_declared_track_keys(consumer_declared, services_info) -> frozenset:
    """Track-view keys of the services a consumer manifest declares.

    `consumer_declared` holds lowercased SOURCE var names (`ray_source`) while
    the track view is keyed by the wizard's service key (`ray_head_source`
    for the multi-container families: Ray, Spark, Airflow, Langfuse, Celery,
    Graph Builder). Matching on the key name dropped those declarations, so a
    wizard-picked track force-disabled them over the manifest's env.values
    (#783). Match on each service's own SOURCE variable instead.
    """
    keys = set()
    for svc in services_info:
        key = svc.key.replace("-", "_") + "_source"
        source_var = (getattr(svc, "env_var_name", "") or "").lower()
        if key in consumer_declared or source_var in consumer_declared:
            keys.add(key)
    return frozenset(keys)


def consumer_override_keys(consumer_declared, services_info) -> frozenset:
    """Track-override keys (normalized folder form, as `start.py` records them
    for `--track`) of the sources a consumer manifest declares. A track picked
    in the wizard skipped and dimmed such a service while the launch kept it
    running (#783 exempts declared sources from the track)."""
    return frozenset(
        normalize_service_key(key.removesuffix("_source").replace("_", "-"))
        for key in consumer_declared_track_keys(consumer_declared, services_info)
    )