clan-core/pkgs/clan-cli/clan_cli/flake.py

import json
import logging
from dataclasses import asdict, dataclass, field
from enum import Enum
from hashlib import sha1
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any

from clan_cli.cmd import Log, RunOpts, run
from clan_cli.dirs import user_cache_dir
from clan_cli.errors import ClanError
from clan_cli.nix import (
    nix_build,
    nix_command,
    nix_config,
    nix_metadata,
    nix_test_store,
)

log = logging.getLogger(__name__)


class SetSelectorType(str, Enum):
    """
    enum for the type of selector in a set.
    For now this is either a string or a maybe selector.
    """

    STR = "str"
    MAYBE = "maybe"


@dataclass
class SetSelector:
    """
    This class represents a selector used in a set.
    type: SetSelectorType = SetSelectorType.STR
    value: str = ""

    a set looks like this:
    {key1,key2}
    """

    type: SetSelectorType = SetSelectorType.STR
    value: str = ""


class SelectorType(str, Enum):
    """
    enum for the type of a selector
    this can be all, string, set or maybe
    """

    ALL = "all"
    STR = "str"
    SET = "set"
    MAYBE = "maybe"


@dataclass
class Selector:
    """
    A class to represent a selector, which selects nix elements one level down.
    consists of a SelectorType and a value.

    if the type is all, no value is needed, since it selects all elements.
    if the type is str, the value is a string, which is the key in a dict.
    if the type is maybe the value is a string, which is the key in a dict.
    if the type is set, the value is a list of SetSelector objects.
    """

    type: SelectorType = SelectorType.STR
    value: str | list[SetSelector] | None = None

    def as_dict(self) -> dict[str, Any]:
        if self.type == SelectorType.SET:
            assert isinstance(self.value, list)
            return {
                "type": self.type.value,
                "value": [asdict(selector) for selector in self.value],
            }
        if self.type == SelectorType.ALL:
            return {"type": self.type.value}
        if self.type == SelectorType.STR:
            assert isinstance(self.value, str)
            return {"type": self.type.value, "value": self.value}
        if self.type == SelectorType.MAYBE:
            assert isinstance(self.value, str)
            return {"type": self.type.value, "value": self.value}
        msg = f"Invalid selector type: {self.type}"
        raise ValueError(msg)


def selectors_as_dict(selectors: list[Selector]) -> list[dict[str, Any]]:
    return [selector.as_dict() for selector in selectors]


def selectors_as_json(selectors: list[Selector]) -> str:
    return json.dumps(selectors_as_dict(selectors))


def parse_selector(selector: str) -> list[Selector]:
    """
    takes a string and returns a list of selectors.

    a selector can be:
    - a string, which is a key in a dict
    - an integer, which is an index in a list
    - a set of strings or integers, which are keys in a dict or indices in a list.
    - the string "*", which selects all elements in a list or dict
    """
    stack: list[str] = []
    selectors: list[Selector] = []
    acc_str: str = ""

    # only used by set for now
    submode = ""
    acc_selectors: list[SetSelector] = []

    for i in range(len(selector)):
        c = selector[i]
        if stack == []:
            mode = "start"
        else:
            mode = stack[-1]

        if mode == "end":
            if c == ".":
                stack.pop()
                if stack != []:
                    msg = "expected empy stack, but got {stack}"
                    raise ValueError(msg)
            else:
                msg = "expected ., but got {c}"
                raise ValueError(msg)

        elif mode == "start":
            if c == "*":
                stack.append("end")
                selectors.append(Selector(type=SelectorType.ALL))
            elif c == "?":
                stack.append("maybe")
            elif c == '"':
                stack += ["str", "quote"]
            elif c == "{":
                stack.append("set")
            elif c == ".":
                selectors.append(Selector(type=SelectorType.STR, value=acc_str))
            else:
                stack.append("str")
                acc_str += c

        elif mode == "set":
            if submode == "" and c == "?":
                submode = "maybe"
            elif c == "\\":
                stack.append("escape")
                if submode == "":
                    submode = "str"
            elif c == '"':
                stack.append("quote")
                if submode == "":
                    submode = "str"
            elif c == ",":
                if submode == "maybe":
                    set_select_type = SetSelectorType.MAYBE
                else:
                    set_select_type = SetSelectorType.STR
                acc_selectors.append(SetSelector(type=set_select_type, value=acc_str))
                submode = ""
                acc_str = ""
            elif c == "}":
                if submode == "maybe":
                    set_select_type = SetSelectorType.MAYBE
                else:
                    set_select_type = SetSelectorType.STR
                acc_selectors.append(SetSelector(type=set_select_type, value=acc_str))
                selectors.append(Selector(type=SelectorType.SET, value=acc_selectors))

                submode = ""
                acc_selectors = []

                acc_str = ""
                stack.pop()
                stack.append("end")
            else:
                acc_str += c
                if submode == "":
                    submode = "str"

        elif mode == "quote":
            if c == '"':
                stack.pop()
            elif c == "\\":
                stack.append("escape")
            else:
                acc_str += c

        elif mode == "escape":
            stack.pop()
            acc_str += c

        elif mode == "str" or mode == "maybe":
            if c == ".":
                stack.pop()
                if mode == "maybe":
                    select_type = SelectorType.MAYBE
                else:
                    select_type = SelectorType.STR
                selectors.append(Selector(type=select_type, value=acc_str))
                acc_str = ""
            elif c == "\\":
                stack.append("escape")
            else:
                acc_str += c

    if stack != []:
        if stack[-1] == "str" or stack[-1] == "maybe":
            if stack[-1] == "maybe":
                select_type = SelectorType.MAYBE
            else:
                select_type = SelectorType.STR
            selectors.append(Selector(type=select_type, value=acc_str))
        elif stack[-1] == "end":
            pass
        else:
            msg = f"expected empty stack, but got {stack}"
            raise ValueError(msg)

    return selectors


@dataclass
class FlakeCacheEntry:
    """
    a recursive structure to store the cache.
    consists of a dict with the keys being the selectors and the values being FlakeCacheEntry objects.

    is_list is used to check if the value is a list.
    exists is used to check if the value exists, which can be false if it was selected via maybe.
    fetched_all is used to check if we have all keys on the current level.
    """

    value: str | float | dict[str, Any] | None = field(default_factory=dict)
    is_list: bool = False
    exists: bool = True
    fetched_all: bool = False

    def insert(
        self,
        value: str | float | dict[str, Any] | list[Any] | None,
        selectors: list[Selector],
    ) -> None:
        selector: Selector
        # if we have no more selectors, it means we select all keys from now one and futher down
        if selectors == []:
            selector = Selector(type=SelectorType.ALL)
        else:
            selector = selectors[0]

        # first we find out if we have all subkeys already

        if self.fetched_all:
            pass
        elif selector.type == SelectorType.ALL:
            self.fetched_all = True

        # if we have a string selector, that means we are usually on a dict or a list, since we cannot walk down scalar values
        # so we passthrough the value to the next level
        if selector.type == SelectorType.STR:
            assert isinstance(selector.value, str)
            assert isinstance(self.value, dict)
            if selector.value not in self.value:
                self.value[selector.value] = FlakeCacheEntry()
            self.value[selector.value].insert(value, selectors[1:])

        # if we get a MAYBE, check if the selector is in the output, if not we create a entry with exists = False
        # otherwise we just insert the value into the current dict
        # we can skip creating the non existing entry if we already fetched all keys
        elif selector.type == SelectorType.MAYBE:
            assert isinstance(self.value, dict)
            assert isinstance(value, dict)
            assert isinstance(selector.value, str)
            if selector.value in value:
                if selector.value not in self.value:
                    self.value[selector.value] = FlakeCacheEntry()
                self.value[selector.value].insert(value[selector.value], selectors[1:])
            elif not self.fetched_all:
                if selector.value not in self.value:
                    self.value[selector.value] = FlakeCacheEntry()
                self.value[selector.value].exists = False

        # insert a dict is pretty straight forward
        elif isinstance(value, dict):
            assert isinstance(self.value, dict)
            for key, value_ in value.items():
                if key not in self.value:
                    self.value[key] = FlakeCacheEntry()
                self.value[key].insert(value_, selectors[1:])

        # to store a list we also use a dict, so we know which indices we have
        elif isinstance(value, list):
            self.is_list = True
            fetched_indices: list[str] = []
            # if we are in a set, we take all the selectors
            if selector.type == SelectorType.SET:
                assert isinstance(selector.value, list)
                for subselector in selector.value:
                    fetched_indices.append(subselector.value)
            # if it's just a str, that is the index
            elif selector.type == SelectorType.STR:
                assert isinstance(selector.value, str)
                fetched_indices = [selector.value]
            # otherwise we just take all the indices, which is the length of the list
            elif selector.type == SelectorType.ALL:
                fetched_indices = list(map(str, range(len(value))))

            # insert is the same is insert a dict
            assert isinstance(self.value, dict)
            for i, requested_index in enumerate(fetched_indices):
                assert isinstance(requested_index, str)
                if requested_index not in self.value:
                    self.value[requested_index] = FlakeCacheEntry()
                self.value[requested_index].insert(value[i], selectors[1:])

        # strings need to be checked if they are store paths
        # if they are, we store them as a dict with the outPath key
        # this is to mirror nix behavior, where the outPath of an attrset is used if no further key is specified
        elif isinstance(value, str) and value.startswith("/nix/store/"):
            assert selectors == []
            if value.startswith("/nix/store/"):
                self.value = {"outPath": FlakeCacheEntry(value)}

        # if we have a normal scalar, we check if it conflicts with a maybe already store value
        # since an empty attrset is the default value, we cannot check that, so we just set it to the value
        elif isinstance(value, float | int | str) or value is None:
            assert selectors == []
            if self.value == {}:
                self.value = value
            elif self.value != value:
                msg = f"Cannot insert {value} into cache, already have {self.value}"
                raise TypeError(msg)

    def is_cached(self, selectors: list[Selector]) -> bool:
        selector: Selector
        if selectors == []:
            return self.fetched_all
        selector = selectors[0]

        # for store paths we have to check if they still exist, otherwise they have to be rebuild and are thus not cached
        if isinstance(self.value, str) and self.value.startswith("/nix/store/"):
            return Path(self.value).exists()

        # if self.value is not dict but we request more selectors, we assume we are cached and an error will be thrown in the select function
        if isinstance(self.value, str | float | int | None):
            return True

        # we just fetch all subkeys, so we need to check of we inserted all keys at this level before
        if selector.type == SelectorType.ALL:
            assert isinstance(self.value, dict)
            if self.fetched_all:
                result = all(
                    self.value[sel].is_cached(selectors[1:]) for sel in self.value
                )
                return result
            return False
        if (
            selector.type == SelectorType.SET
            and isinstance(selector.value, list)
            and isinstance(self.value, dict)
        ):
            for requested_selector in selector.value:
                val = requested_selector.value
                if val not in self.value:
                    # if we fetched all keys and we are not in the dict, we can assume we are cached
                    return self.fetched_all
                # if a key does not exist from a previous fetch, we can assume it is cached
                if self.value[val].exists is False:
                    return True
                if not self.value[val].is_cached(selectors[1:]):
                    return False

            return True

        # string and maybe work the same for cache checking
        if (
            selector.type == SelectorType.STR or selector.type == SelectorType.MAYBE
        ) and isinstance(self.value, dict):
            assert isinstance(selector.value, str)
            val = selector.value
            if val not in self.value:
                # if we fetched all keys and we are not in there, refetching won't help, so we can assume we are cached
                return self.fetched_all
            if self.value[val].exists is False:
                return True
            return self.value[val].is_cached(selectors[1:])

        return False

    def select(self, selectors: list[Selector]) -> Any:
        selector: Selector
        if selectors == []:
            selector = Selector(type=SelectorType.ALL)
        else:
            selector = selectors[0]

        # mirror nix behavior where we return outPath if no further selector is specified
        if selectors == [] and isinstance(self.value, dict) and "outPath" in self.value:
            return self.value["outPath"].value

        # if we are at the end of the selector chain, we return the value
        if selectors == [] and isinstance(self.value, str | float | int | None):
            return self.value

        # if we fetch a specific key, we return the recurse into that value in the dict
        if selector.type == SelectorType.STR and isinstance(self.value, dict):
            assert isinstance(selector.value, str)
            return self.value[selector.value].select(selectors[1:])

        # if we are a MAYBE selector, we check if the key exists in the dict
        if selector.type == SelectorType.MAYBE and isinstance(self.value, dict):
            assert isinstance(selector.value, str)
            if selector.value in self.value:
                if self.value[selector.value].exists:
                    return {
                        selector.value: self.value[selector.value].select(selectors[1:])
                    }
                return {}
            if self.fetched_all:
                return {}

        # otherwise we return a list or a dict
        if isinstance(self.value, dict):
            keys_to_select: list[str] = []
            # if we want to select all keys, we take all existing sub elements
            if selector.type == SelectorType.ALL:
                for key in self.value:
                    if self.value[key].exists:
                        keys_to_select.append(key)

            # if we want to select a set of keys, we take the keys from the selector
            if selector.type == SelectorType.SET:
                assert isinstance(selector.value, list)
                for subselector in selector.value:
                    # make sure the keys actually exist if we have a maybe selector
                    if subselector.type == SetSelectorType.MAYBE:
                        if (
                            subselector.value in self.value
                            and self.value[subselector.value].exists
                        ):
                            keys_to_select.append(subselector.value)
                    else:
                        keys_to_select.append(subselector.value)

            # if we are a list, return a list
            if self.is_list:
                result = []
                for index in keys_to_select:
                    result.append(self.value[index].select(selectors[1:]))
                    return result

            # otherwise return a dict
            return {k: self.value[k].select(selectors[1:]) for k in keys_to_select}

        # return a KeyError if we cannot fetch the key
        str_selector: str
        if selector.type == SelectorType.ALL:
            str_selector = "*"
        elif selector.type == SelectorType.SET:
            subselectors: list[str] = []
            assert isinstance(selector.value, list)
            for subselector in selector.value:
                subselectors.append(subselector.value)
            str_selector = "{" + ",".join(subselectors) + "}"
        else:
            assert isinstance(selector.value, str)
            str_selector = selector.value

        raise KeyError(str_selector)

    def __getitem__(self, name: str) -> "FlakeCacheEntry":
        if isinstance(self.value, dict):
            return self.value[name]
        raise KeyError(name)

    def as_json(self) -> dict[str, Any]:
        json_data: Any = {}
        if isinstance(self.value, dict):
            value = json_data["value"] = {}
            for k, v in self.value.items():
                value[k] = v.as_json()
        else:  # == str | float | None
            json_data["value"] = self.value

        json_data["is_list"] = self.is_list
        json_data["exists"] = self.exists
        json_data["fetched_all"] = self.fetched_all
        return json_data

    @staticmethod
    def from_json(json_data: dict[str, Any]) -> "FlakeCacheEntry":
        raw_value = json_data.get("value")
        if isinstance(raw_value, dict):
            value: Any = {}
            for k, v in raw_value.items():
                value[k] = FlakeCacheEntry.from_json(v)
        else:  # == str | float | None
            value = raw_value

        is_list = json_data.get("is_list", False)
        exists = json_data.get("exists", True)
        fetched_all = json_data.get("fetched_all", False)

        entry = FlakeCacheEntry(
            value=value, is_list=is_list, exists=exists, fetched_all=fetched_all
        )
        return entry

    def __repr__(self) -> str:
        if isinstance(self.value, dict):
            return f"FlakeCache {{{', '.join([str(k) for k in self.value])}}}"
        return f"FlakeCache {self.value}"


@dataclass
class FlakeCache:
    """
    an in-memory cache for flake outputs, uses a recursive FLakeCacheEntry structure
    """

    def __init__(self) -> None:
        self.cache: FlakeCacheEntry = FlakeCacheEntry()

    def insert(self, data: dict[str, Any], selector_str: str) -> None:
        if selector_str:
            selectors = parse_selector(selector_str)
        else:
            selectors = []

        self.cache.insert(data, selectors)

    def select(self, selector_str: str) -> Any:
        selectors = parse_selector(selector_str)
        return self.cache.select(selectors)

    def is_cached(self, selector_str: str) -> bool:
        selectors = parse_selector(selector_str)
        return self.cache.is_cached(selectors)

    def save_to_file(self, path: Path) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        with NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as temp_file:
            data = {"cache": self.cache.as_json()}
            json.dump(data, temp_file)
            temp_file.close()
            Path(temp_file.name).rename(path)

    def load_from_file(self, path: Path) -> None:
        with path.open("r") as f:
            log.debug(f"Loading cache from {path}")
            data = json.load(f)
            self.cache = FlakeCacheEntry.from_json(data["cache"])


@dataclass
class Flake:
    """
    This class represents a flake, and is used to interact with it.
    values can be accessed using the select method, which will fetch the value from the cache if it is present.
    """

    identifier: str
    inputs_from: str | None = None
    hash: str | None = None
    flake_cache_path: Path | None = None
    store_path: str | None = None
    cache: FlakeCache | None = None
    _cache: FlakeCache | None = None
    _path: Path | None = None
    _is_local: bool | None = None

    @classmethod
    def from_json(cls: type["Flake"], data: dict[str, Any]) -> "Flake":
        return cls(data["identifier"])

    def __str__(self) -> str:
        return self.identifier

    def __hash__(self) -> int:
        return hash(self.identifier)

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Flake):
            return NotImplemented
        return self.identifier == other.identifier

    @property
    def is_local(self) -> bool:
        if self._is_local is None:
            self.invalidate_cache()
        assert isinstance(self._is_local, bool)
        return self._is_local

    @property
    def path(self) -> Path:
        if self._path is None:
            self.invalidate_cache()
        assert isinstance(self._path, Path)
        return self._path

    def load_cache(self) -> None:
        path = self.flake_cache_path
        if path is None or self._cache is None or not path.exists():
            return
        try:
            self._cache.load_from_file(path)
        except Exception as e:
            log.warning(f"Failed load eval cache: {e}. Continue without cache")

    def invalidate_cache(self) -> None:
        """
        Invalidate the cache and reload it.

        This method is used to refresh the cache by reloading it from the flake.
        """
        cmd = [
            "flake",
            "prefetch",
            "--json",
            "--option",
            "flake-registry",
            "",
            self.identifier,
        ]

        if self.inputs_from:
            cmd += ["--inputs-from", self.inputs_from]

        flake_prefetch = run(nix_command(cmd))
        flake_metadata = json.loads(flake_prefetch.stdout)
        self.store_path = flake_metadata["storePath"]
        self.hash = flake_metadata["hash"]

        self._cache = FlakeCache()
        assert self.hash is not None
        hashed_hash = sha1(self.hash.encode()).hexdigest()
        self.flake_cache_path = (
            Path(user_cache_dir()) / "clan" / "flakes-v2" / hashed_hash
        )
        self.load_cache()

        if "original" not in flake_metadata:
            flake_metadata = nix_metadata(self.identifier)

        if flake_metadata["original"].get("url", "").startswith("file:"):
            self._is_local = True
            path = flake_metadata["original"]["url"].removeprefix("file://")
            path = path.removeprefix("file:")
            self._path = Path(path)
        elif flake_metadata["original"].get("path"):
            self._is_local = True
            self._path = Path(flake_metadata["original"]["path"])
        else:
            self._is_local = False
            assert self.store_path is not None
            self._path = Path(self.store_path)

    def get_from_nix(
        self,
        selectors: list[str],
        nix_options: list[str] | None = None,
    ) -> None:
        """
        Retrieves specific attributes from a Nix flake using the provided selectors.

        This function interacts with the Nix build system to fetch and process
        attributes from a flake. It uses the provided selectors to determine which
        attributes to retrieve and optionally accepts additional Nix options for
        customization. The results are cached for future use.
        Used mostly as a lowlevel function for `precache` and `select` methods.

        Args:
            selectors (list[str]): A list of attribute selectors to fetch from the flake.
            nix_options (list[str] | None): Optional additional options to pass to the Nix build command.

        Raises:
            ClanError: If the number of outputs does not match the number of selectors.
            AssertionError: If the cache or flake cache path is not properly initialized.
        """
        if self._cache is None:
            self.invalidate_cache()
        assert self._cache is not None

        if nix_options is None:
            nix_options = []

        str_selectors: list[str] = []
        for selector in selectors:
            str_selectors.append(selectors_as_json(parse_selector(selector)))

        config = nix_config()
        nix_code = f"""
            let
              flake = builtins.getFlake("path:{self.store_path}?narHash={self.hash}");
            in
              flake.inputs.nixpkgs.legacyPackages.{config["system"]}.writeText "clan-flake-select" (
                builtins.toJSON [ {" ".join([f"(flake.clanInternals.clanLib.select.applySelectors (builtins.fromJSON ''{attr}'') flake)" for attr in str_selectors])} ]
              )
        """
        if tmp_store := nix_test_store():
            nix_options += ["--store", str(tmp_store)]
            nix_options.append("--impure")

        build_output = Path(
            run(
                nix_build(["--expr", nix_code, *nix_options]), RunOpts(log=Log.NONE)
            ).stdout.strip()
        )

        if tmp_store:
            build_output = tmp_store.joinpath(*build_output.parts[1:])
        outputs = json.loads(build_output.read_text())
        if len(outputs) != len(selectors):
            msg = f"flake_prepare_cache: Expected {len(outputs)} outputs, got {len(outputs)}"
            raise ClanError(msg)
        self.load_cache()
        for i, selector in enumerate(selectors):
            self._cache.insert(outputs[i], selector)
        if self.flake_cache_path:
            self._cache.save_to_file(self.flake_cache_path)

    def precache(
        self,
        selectors: list[str],
        nix_options: list[str] | None = None,
    ) -> None:
        """
        Ensures that the specified selectors are cached locally.

        This function checks if the given selectors are already cached. If not, it
        fetches them using the Nix build system and stores them in the local cache.
        It ensures that the cache is initialized before performing these operations.

        Args:
            selectors (list[str]): A list of attribute selectors to check and cache.
            nix_options (list[str] | None): Optional additional options to pass to the Nix build command.
        """
        if self._cache is None:
            self.invalidate_cache()
        assert self._cache is not None
        assert self.flake_cache_path is not None
        not_fetched_selectors = []
        for selector in selectors:
            if not self._cache.is_cached(selector):
                not_fetched_selectors.append(selector)
        if not_fetched_selectors:
            self.get_from_nix(not_fetched_selectors, nix_options)

    def select(
        self,
        selector: str,
        nix_options: list[str] | None = None,
    ) -> Any:
        """
        Selects a value from the cache based on the provided selector string.
        Fetches it via nix_build if it is not already cached.

        Args:
            selector (str): The attribute selector string to fetch the value for.
            nix_options (list[str] | None): Optional additional options to pass to the Nix build command.
        """
        if self._cache is None:
            self.invalidate_cache()
        assert self._cache is not None
        assert self.flake_cache_path is not None

        if not self._cache.is_cached(selector):
            log.debug(f"Cache miss for {selector}")
            self.get_from_nix([selector], nix_options)
        value = self._cache.select(selector)
        return value