headquarter/apps/api/src/services/config/config_profile_resolver.py

"""Config profile resolver service.

Provides recursive ordered include resolution with deterministic merge rules
and cycle protection.
"""

import logging
import os
import uuid
from dataclasses import dataclass, field
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models import ConfigProfile, ConfigProfileInclude

logger = logging.getLogger(__name__)


class ConfigProfileCycleError(Exception):
    """Raised when a cycle is detected in profile includes."""

    pass


class ConfigProfileNotFoundError(Exception):
    """Raised when a referenced profile is not found."""

    pass


@dataclass
class ResolvedMount:
    """A resolved mount with merged files and final mode."""

    target: str
    mode: str
    files: dict[str, str] = field(default_factory=dict)
    overridden_files: dict[str, str] = field(default_factory=dict)


@dataclass
class ResolvedProfile:
    """The fully resolved output of a config profile."""

    profile_id: uuid.UUID
    profile_name: str
    env_vars: dict[str, str] = field(default_factory=dict)
    runtime_hints: dict[str, Any] = field(default_factory=dict)
    mounts: dict[str, ResolvedMount] = field(default_factory=dict)
    git_mounts: list[dict[str, Any]] = field(default_factory=list)
    files: dict[str, str] = field(default_factory=dict)
    env_overrides: dict[str, str] = field(default_factory=dict)
    hint_overrides: dict[str, str] = field(default_factory=dict)
    file_overrides: dict[str, str] = field(default_factory=dict)
    mount_overrides: dict[str, str] = field(default_factory=dict)
    included_profiles: list[dict[str, Any]] = field(default_factory=list)


def _detect_cycle(
    profile_id: uuid.UUID, visited: set[uuid.UUID], path: list[uuid.UUID]
) -> bool:
    """Detect if adding profile_id to path would create a cycle.

    Args:
        profile_id: The profile ID to check.
        visited: Set of already-visited profile IDs in current resolution.
        path: Current resolution path for error reporting.

    Returns:
        True if a cycle would be created.
    """
    if profile_id in visited:
        return True
    return False


def _merge_env_vars(
    base: dict[str, str],
    overlay: dict[str, str],
    overrides: dict[str, str],
    source_name: str,
) -> dict[str, str]:
    """Merge env vars, tracking overrides.

    Later values replace earlier values.
    """
    result = dict(base)
    for key, value in overlay.items():
        if key in result and result[key] != value:
            overrides[key] = source_name
        result[key] = value
    return result


def _merge_runtime_hints(
    base: dict[str, Any],
    overlay: dict[str, Any],
    overrides: dict[str, str],
    source_name: str,
) -> dict[str, Any]:
    """Merge runtime hints, tracking overrides.

    Later values replace earlier values.
    """
    result = dict(base)
    for key, value in overlay.items():
        if key in result and result[key] != value:
            overrides[key] = source_name
        result[key] = value
    return result


def _merge_files(
    base: dict[str, str],
    overlay: dict[str, str],
    overrides: dict[str, str],
    source_name: str,
) -> dict[str, str]:
    """Merge file maps, tracking overrides.

    Later relative file paths win.
    """
    result = dict(base)
    for path, content in overlay.items():
        if path in result and result[path] != content:
            overrides[path] = source_name
        result[path] = content
    return result


def _merge_mounts(
    base: dict[str, ResolvedMount],
    overlay: list[dict[str, Any]],
    overrides: dict[str, str],
    source_name: str,
) -> dict[str, ResolvedMount]:
    """Merge mounts, tracking overrides.

    Mounts with the same target path have their file maps merged and later
    relative file paths win. Mode conflicts: later layer wins.
    """
    result = dict(base)
    for mount_data in overlay:
        target = mount_data["target"]
        mode = mount_data.get("mode", "rw")
        files = mount_data.get("files", {})

        if target in result:
            existing = result[target]
            merged_files = dict(existing.files)
            file_overrides = dict(existing.overridden_files)
            for rel_path, content in files.items():
                if rel_path in merged_files and merged_files[rel_path] != content:
                    file_overrides[rel_path] = source_name
                merged_files[rel_path] = content
            if existing.mode != mode:
                overrides[target] = source_name
            result[target] = ResolvedMount(
                target=target,
                mode=mode,
                files=merged_files,
                overridden_files=file_overrides,
            )
        else:
            result[target] = ResolvedMount(
                target=target,
                mode=mode,
                files=dict(files),
            )
    return result


def _find_mount_conflicts(
    profile: ConfigProfile,
    resolved: ResolvedProfile,
) -> list[dict[str, Any]]:
    """Find mounts on the profile that override mounts from included profiles.

    Returns a list of conflict descriptors with the target path, the included
    profile that originally provided the mount, and the current profile name.
    """
    conflicts = []
    own_targets = {m["target"] for m in (profile.mounts or [])}
    own_files = {f for m in (profile.mounts or []) for f in m.get("files", {}).keys()}

    for mount in resolved.mounts.values():
        if mount.target in own_targets:
            # The profile itself has a mount at the same target as an included one.
            conflicts.append(
                {
                    "type": "mount_target",
                    "target": mount.target,
                    "overridden_by": profile.name,
                    "source": resolved.profile_name,
                }
            )
            continue

        for rel_path in mount.files:
            if rel_path in own_files:
                conflicts.append(
                    {
                        "type": "mount_file",
                        "target": mount.target,
                        "file": rel_path,
                        "overridden_by": profile.name,
                        "source": resolved.profile_name,
                    }
                )

    return conflicts


def _merge_git_mounts(
    base: list[dict[str, Any]],
    overlay: list[dict[str, Any]],
    source_name: str,
) -> list[dict[str, Any]]:
    """Merge git mounts from included profiles.

    Entries with the same remote_url + branch have their mappings concatenated.
    Different repos are kept as separate entries.
    All entries are normalized to the mappings format.
    """
    result = list(base)
    # Normalize existing entries to mappings format
    for i, m in enumerate(result):
        result[i] = _normalize_git_mount_entry(dict(m))

    # Build lookup by (remote_url, branch)
    seen = {}
    for i, m in enumerate(result):
        key = (m["remote_url"], m.get("branch"))
        seen[key] = i

    for mount in overlay:
        mount = _normalize_git_mount_entry(dict(mount))
        key = (mount["remote_url"], mount.get("branch"))
        if key in seen:
            # Same repo+branch: concatenate mappings, dedup by (source_path, target_path)
            existing = result[seen[key]]
            existing_sources = {
                (m["source_path"], m["target_path"])
                for m in existing.get("mappings", [])
            }
            for mapping in mount.get("mappings", []):
                map_key = (mapping["source_path"], mapping["target_path"])
                if map_key not in existing_sources:
                    existing["mappings"].append(dict(mapping))
                    existing_sources.add(map_key)
        else:
            seen[key] = len(result)
            result.append(mount)
    return result


def _normalize_git_mount_entry(entry: dict[str, Any]) -> dict[str, Any]:
    """Normalize a git mount entry to the unified mappings format.

    Converts legacy source_path + target_path into a single-entry mappings array.
    """
    entry = dict(entry)
    if "mappings" not in entry or not entry.get("mappings"):
        source = entry.get("source_path", ".")
        target = entry.get("target_path")
        if target is not None:
            entry["mappings"] = [{"source_path": source, "target_path": target}]
    # Remove legacy fields once normalized
    entry.pop("source_path", None)
    entry.pop("target_path", None)
    return entry


async def _resolve_profile_recursive(
    session: AsyncSession,
    profile_id: uuid.UUID,
    visited: set[uuid.UUID],
    path: list[uuid.UUID],
) -> ResolvedProfile:
    """Recursively resolve a profile and its includes.

    Args:
        session: Database session.
        profile_id: Profile ID to resolve.
        visited: Set of already-visited profile IDs in current resolution chain.
        path: Current resolution path for error reporting.

    Returns:
        ResolvedProfile with all includes merged.

    Raises:
        ConfigProfileCycleError: If a cycle is detected.
        ConfigProfileNotFoundError: If the profile is not found.
    """
    if _detect_cycle(profile_id, visited, path):
        cycle_path = " -> ".join(str(p) for p in path + [profile_id])
        raise ConfigProfileCycleError(
            f"Cycle detected in profile includes: {cycle_path}"
        )

    profile = await session.get(ConfigProfile, profile_id)
    if profile is None:
        raise ConfigProfileNotFoundError(f"Config profile not found: {profile_id}")

    new_visited = visited | {profile_id}
    new_path = path + [profile_id]

    result = ResolvedProfile(
        profile_id=profile.id,
        profile_name=profile.name,
    )

    # Resolve includes in order
    include_query = (
        select(ConfigProfileInclude)
        .where(ConfigProfileInclude.profile_id == profile_id)
        .order_by(ConfigProfileInclude.order_index)
    )
    include_result = await session.execute(include_query)
    includes = include_result.scalars().all()

    for include in includes:
        included = await _resolve_profile_recursive(
            session, include.included_profile_id, new_visited, new_path
        )
        result.included_profiles.append(
            {
                "id": str(included.profile_id),
                "name": included.profile_name,
            }
        )

        result.env_vars = _merge_env_vars(
            result.env_vars,
            included.env_vars,
            result.env_overrides,
            included.profile_name,
        )
        result.runtime_hints = _merge_runtime_hints(
            result.runtime_hints,
            included.runtime_hints,
            result.hint_overrides,
            included.profile_name,
        )
        result.files = _merge_files(
            result.files, included.files, result.file_overrides, included.profile_name
        )
        result.mounts = _merge_mounts(
            result.mounts,
            [
                {"target": m.target, "mode": m.mode, "files": m.files}
                for m in included.mounts.values()
            ],
            result.mount_overrides,
            included.profile_name,
        )
        result.git_mounts = _merge_git_mounts(
            result.git_mounts, included.git_mounts, included.profile_name
        )

    # Apply the profile's own settings (selected profile overrides includes)
    result.env_vars = _merge_env_vars(
        result.env_vars,
        profile.env_vars or {},
        result.env_overrides,
        profile.name,
    )
    result.runtime_hints = _merge_runtime_hints(
        result.runtime_hints,
        profile.runtime_hints or {},
        result.hint_overrides,
        profile.name,
    )
    result.files = _merge_files(
        result.files,
        profile.files or {},
        result.file_overrides,
        profile.name,
    )
    result.mounts = _merge_mounts(
        result.mounts,
        profile.mounts or [],
        result.mount_overrides,
        profile.name,
    )
    result.git_mounts = _merge_git_mounts(
        result.git_mounts,
        profile.git_mounts or [],
        profile.name,
    )
    return result


async def resolve_profile(
    session: AsyncSession,
    profile_id: uuid.UUID,
) -> ResolvedProfile:
    """Resolve a config profile with all includes.

    Args:
        session: Database session.
        profile_id: Profile ID to resolve.

    Returns:
        ResolvedProfile with merged env vars, runtime hints, mounts, and files.

    Raises:
        ConfigProfileCycleError: If a cycle is detected in includes.
        ConfigProfileNotFoundError: If the profile is not found.
    """
    return await _resolve_profile_recursive(session, profile_id, set(), [])


async def check_include_cycle(
    session: AsyncSession,
    profile_id: uuid.UUID,
    new_include_id: uuid.UUID | None = None,
) -> list[uuid.UUID] | None:
    """Check if adding an include would create a cycle.

    Used at save time to validate include relationships before persisting.

    Args:
        session: Database session.
        profile_id: The profile that would receive the new include.
        new_include_id: Optional new profile to include. If None, checks existing includes.

    Returns:
        The cycle path as a list of UUIDs if a cycle exists, otherwise None.
    """

    async def _check_from(
        current_id: uuid.UUID,
        target_id: uuid.UUID,
        visited: set[uuid.UUID],
        path: list[uuid.UUID],
    ) -> list[uuid.UUID] | None:
        if current_id in visited:
            if current_id == target_id:
                return path + [current_id]
            return None
        if current_id == target_id and path:
            return path + [current_id]

        new_visited = visited | {current_id}
        new_path = path + [current_id]

        include_query = (
            select(ConfigProfileInclude)
            .where(ConfigProfileInclude.profile_id == current_id)
            .order_by(ConfigProfileInclude.order_index)
        )
        include_result = await session.execute(include_query)
        includes = include_result.scalars().all()

        for include in includes:
            cycle = await _check_from(
                include.included_profile_id, target_id, new_visited, new_path
            )
            if cycle is not None:
                return cycle
        return None

    # Check if new_include_id can reach profile_id (would create cycle)
    if new_include_id is not None:
        cycle = await _check_from(new_include_id, profile_id, set(), [])
        if cycle is not None:
            return cycle

    # Also check existing includes for cycles
    cycle = await _check_from(profile_id, profile_id, set(), [])
    if cycle is not None and len(cycle) > 1:
        return cycle

    return None


def apply_resolved_profile(
    instance_dir: str,
    resolved: ResolvedProfile,
    home_dir: str = "/root",
) -> tuple[dict[str, str], dict[str, str], list[dict], dict[str, Any]]:
    """Apply a resolved profile to an instance directory.

    Stages files, writes env vars, and prepares mount volumes.

    Args:
        instance_dir: Path to the instance directory.
        resolved: The resolved profile.

    Returns:
        Tuple of (env_vars, files, volume_mounts, runtime_hints).
        env_vars: Merged environment variables.
        files: Relative file paths to content for the instance.
        volume_mounts: List of Docker volume mount dicts.
        runtime_hints: Extracted runtime hints.
    """
    from pathlib import Path

    instance_path = Path(instance_dir)
    env_vars = dict(resolved.env_vars)
    files = dict(resolved.files)
    volume_mounts = []

    # Write profile files to instance directory
    for file_path, content in files.items():
        full_path = instance_path / file_path
        try:
            full_path.resolve().relative_to(instance_path.resolve())
        except ValueError:
            logger.warning(
                "Profile file path escapes instance directory: %s", file_path
            )
            continue
        full_path.parent.mkdir(parents=True, exist_ok=True)
        full_path.write_text(content)

    # Stage mount directories and prepare directory-level volume mounts.
    # Each ResolvedMount targets a container directory; we stage all of its
    # files under a single host directory and bind-mount that directory. This
    # keeps the target directory writable by the container user, instead of
    # having Docker create a root-owned parent directory when only individual
    # files are mounted.
    for mount in resolved.mounts.values():
        if not mount.files:
            continue

        expanded_target = os.path.normpath(
            expand_container_path(mount.target, home_dir)
        )
        mount_dir = (
            instance_path / "mounts" / expanded_target.lstrip("/").replace("/", "_")
        )
        mount_dir.mkdir(parents=True, exist_ok=True)

        for file_path, content in mount.files.items():
            full_path = mount_dir / file_path
            try:
                full_path.resolve().relative_to(mount_dir.resolve())
            except ValueError:
                logger.warning("Mount file path escapes mount directory: %s", file_path)
                continue
            full_path.parent.mkdir(parents=True, exist_ok=True)
            full_path.write_text(content)

        volume_mounts.append(
            {
                "source": str(mount_dir),
                "target": expanded_target,
                "type": "bind",
                "readonly": mount.mode in ("ro", "readonly"),
            }
        )

    return env_vars, files, volume_mounts, resolved.runtime_hints


def expand_container_path(path: str, home_dir: str) -> str:
    """Expand ~ and $HOME in a container path to the actual home directory.

    Only expands at the start of the path (e.g., ~/foo, $HOME/foo, $HOME).
    Leaves mid-string occurrences unchanged.

    Args:
        path: Container path that may contain ~ or $HOME.
        home_dir: The container's home directory (e.g., /home/user or /root).

    Returns:
        Path with ~ and $HOME expanded.
    """
    if path.startswith("~/"):
        return os.path.join(home_dir, path[2:])
    if path == "~":
        return home_dir
    if path.startswith("$HOME/"):
        return home_dir + "/" + path[6:]
    if path == "$HOME":
        return home_dir
    return path


def resolved_profile_to_dict(resolved: ResolvedProfile) -> dict[str, Any]:
    """Convert a ResolvedProfile to a plain dict for serialization.

    Args:
        resolved: The resolved profile.

    Returns:
        Dict with env_vars, runtime_hints, mounts, files, and metadata.
    """
    return {
        "profile_id": str(resolved.profile_id),
        "profile_name": resolved.profile_name,
        "env_vars": resolved.env_vars,
        "runtime_hints": resolved.runtime_hints,
        "mounts": [
            {
                "target": m.target,
                "mode": m.mode,
                "files": m.files,
                "overridden_files": m.overridden_files,
            }
            for m in resolved.mounts.values()
        ],
        "files": resolved.files,
        "overrides": {
            "env_vars": resolved.env_overrides,
            "runtime_hints": resolved.hint_overrides,
            "files": resolved.file_overrides,
            "mounts": resolved.mount_overrides,
        },
        "git_mounts": resolved.git_mounts,
        "included_profiles": resolved.included_profiles,
    }