computer-playbook/plugins/lookup/compose_volumes.py

"""Lookup `compose_volumes`: render the top-level `volumes:` / `configs:` /
`secrets:` block. Auto-wires `application_id`, the `applications` registry,
DEPLOYMENT_MODE, and the `storage` mapping from the templating context.
"""

from __future__ import annotations

import contextlib
from typing import Any

from ansible.errors import AnsibleError, AnsibleFilterError
from ansible.plugins.loader import lookup_loader
from ansible.plugins.lookup import LookupBase

from plugins.filter.docker.service_enabled import (
    FilterModule as _DockerServiceEnabledFilter,
)
from plugins.filter.get.entity_name import get_entity_name

# nocheck: lookup-cache-import (raw-volume accessor: canonical meta/volumes.yml shape)
from utils.cache.applications import get_canonical_volumes
from utils.cache.yaml import dump_yaml_str
from utils.roles.applications.config import get
from utils.roles.applications.mounts import (
    content_hash,
    mount_when_passes,
    normalize_volumes_meta,
)
from utils.roles.applications.services.database import (
    REALIGN_CONFIG_KEY,
    get_database_service_config,
    resolve_database_service_key,
)
from utils.roles.applications.services.sso import get_sso_config
from utils.storage.nfs import swarm_nfs_backed
from utils.templating.ansible import _trust_as_template, to_plain


def _resolve_database_volume_name(
    applications: dict[str, Any], application_id: str, dbtype: str
) -> str:
    consumer_entity = get_entity_name(application_id)
    db_id = f"svc-db-{dbtype}"
    central_name = get(
        applications=applications,
        application_id=db_id,
        config_path=f"services.{dbtype}.name",
        strict=False,
        default="",
        skip_missing_app=True,
    )
    central_name = (str(central_name) if central_name is not None else "").strip()
    service_cfg = get_database_service_config(applications, application_id)
    central_enabled = bool(service_cfg.get("shared", False))
    host = central_name if central_enabled else "database"
    volume_prefix = "" if central_enabled else f"{consumer_entity}_"
    return f"{volume_prefix}{host}"


def _swarm_nfs_driver_opts(dir_var_lib: str, volume_name: str) -> dict[str, Any]:
    return {
        "driver": "local",
        "driver_opts": {
            "type": "none",
            "o": "bind",
            "device": f"{dir_var_lib.rstrip('/')}/{volume_name}",
        },
    }


def _read_file_for_hash(source: str) -> str:
    """Read a file's content for hash-based naming. Returns the path as a
    stable fallback when the file isn't materialised yet (lint runs)."""
    try:
        from utils.cache.files import read_text

        return read_text(source)
    except (OSError, ValueError):
        return source


def _config_secret_name(role_entity: str, user_name: str, source: str) -> str:
    """Build a swarm-rotation-safe name: ``{role}_{name}_{sha8(content)}``."""
    digest = content_hash(_read_file_for_hash(source))
    return f"{role_entity}_{user_name}_{digest}"


def _maybe_render(value: Any, render_jinja: Any) -> str:
    text = str(value)
    if render_jinja is None or ("{{" not in text and "{%" not in text):
        return text
    try:
        return str(render_jinja(text))
    except AnsibleError:
        return text


def compose_volumes(
    applications: dict[str, Any],
    application_id: str,
    *,
    extra_volumes: dict[str, dict[str, Any]] | None = None,
    extra_configs: dict[str, dict[str, Any]] | None = None,
    extra_secrets: dict[str, dict[str, Any]] | None = None,
    deployment_mode: str = "compose",
    storage: dict[str, Any] | None = None,
    dir_var_lib: str,
    render_jinja: Any = None,
) -> str:
    """Render the top-level ``volumes:`` / ``configs:`` / ``secrets:`` block.

    Reads the canonical dict-of-dicts shape from
    ``roles/<role>/meta/volumes.yml`` (YAML key = semantic short name;
    optional ``name:`` field = docker volume name). Entries with
    ``type: config`` / ``type: secret`` emit their respective top-level
    sections; everything else flows through the volumes section.

    A ``type: volume`` entry may set ``nfs: false`` to opt out of the
    swarm NFS rewrite: the volume then stays a plain node-local named
    volume even when ``storage.backend`` is ``nfs`` (e.g. gitaly
    repository storage, which upstream forbids on NFS). Any other
    ``nfs`` value (absent, ``true``, or the uid/gid/mode dict consumed
    by the NFS subdir pre-creation) keeps the rewrite.
    """

    if applications is None:
        raise AnsibleFilterError("compose_volumes: 'applications' must not be None")
    if not isinstance(applications, dict):
        raise AnsibleFilterError("compose_volumes: 'applications' must be a dict")
    if not application_id or not isinstance(application_id, str):
        raise AnsibleFilterError(
            "compose_volumes: 'application_id' must be a non-empty string"
        )
    if application_id not in applications:
        raise AnsibleFilterError(
            f"compose_volumes: unknown application_id '{application_id}'"
        )

    role_entity = get_entity_name(application_id)
    volumes: dict[str, Any] = {}
    configs: dict[str, Any] = {}
    secrets: dict[str, Any] = {}

    try:
        database_service_key = resolve_database_service_key(
            applications, application_id
        )
    except ValueError as exc:
        raise AnsibleFilterError(
            "compose_volumes: "
            f"{exc}. Simultaneous postgres + mariadb on the same role "
            "is not supported (the embedded service templates collide "
            "on the `database` service key, host name, and volume "
            "key); pick one dbtype per role."
        ) from exc
    database_service = get_database_service_config(applications, application_id)
    database_needed = bool(database_service_key) and not bool(
        database_service.get("shared", False)
    )

    if database_needed:
        volumes["database"] = {
            "name": _resolve_database_volume_name(
                applications, application_id, database_service_key
            )
        }

    for engine in ("seaweedfs", "minio"):
        engine_enabled = bool(
            get(
                applications=applications,
                application_id=application_id,
                config_path=f"services.{engine}.enabled",
                strict=False,
                default=False,
            )
        )
        engine_shared = bool(
            get(
                applications=applications,
                application_id=application_id,
                config_path=f"services.{engine}.shared",
                strict=False,
                default=False,
            )
        )
        if engine_enabled and not engine_shared:
            volumes[engine] = {"name": f"{get_entity_name(application_id)}_{engine}"}

    sso = get_sso_config(applications, application_id)

    if (
        _DockerServiceEnabledFilter.is_docker_service_enabled(
            applications, application_id, "redis"
        )
        or sso["is_proxy_gated"]
    ):
        volumes["redis"] = {"name": f"{role_entity}_redis"}

    if extra_volumes:
        volumes.update(extra_volumes)
    for section, extra in ((configs, extra_configs), (secrets, extra_secrets)):
        for key, spec in (extra or {}).items():
            named = spec
            if isinstance(spec, dict) and spec.get("file") and not spec.get("name"):
                named = {
                    **spec,
                    "name": _config_secret_name(role_entity, key, str(spec["file"])),
                }
            section[key] = named

    role_data = applications.get(application_id) or {}
    raw_meta_volumes = (
        get_canonical_volumes(application_id) or role_data.get("volumes") or {}
    )
    canonical_entries = normalize_volumes_meta(raw_meta_volumes)

    for semantic_name, entry in canonical_entries.items():
        vtype = entry.get("type", "volume")

        if vtype == "volume":
            if semantic_name in volumes:
                continue
            spec: dict[str, Any] = {}
            docker_name = entry.get("name")
            if docker_name:
                spec["name"] = docker_name
            if "nfs" in entry:
                spec["nfs"] = entry["nfs"]
            volumes[semantic_name] = spec
            continue

        if vtype in ("config", "secret") and not any(
            mount_when_passes(mount, render_jinja)
            for mount in entry.get("mounts") or []
            if isinstance(mount, dict)
        ):
            continue

        if vtype == "config":
            source = _maybe_render(entry.get("source", ""), render_jinja)
            configs[semantic_name] = {
                "name": _maybe_render(entry.get("name", ""), render_jinja)
                or _config_secret_name(role_entity, semantic_name, source),
                "file": source,
            }
            continue

        if vtype == "secret":
            source = _maybe_render(entry.get("source", ""), render_jinja)
            secrets[semantic_name] = {
                "name": _maybe_render(entry.get("name", ""), render_jinja)
                or _config_secret_name(role_entity, semantic_name, source),
                "file": source,
            }
            continue

    storage_backend = (storage or {}).get("backend", "local")

    for vol_name, vol_spec in list(volumes.items()):
        if not isinstance(vol_spec, dict):
            continue
        nfs_meta = vol_spec.pop("nfs", None)
        if swarm_nfs_backed(
            {"nfs": nfs_meta},
            application_id=application_id,
            deployment_mode=deployment_mode,
            storage_backend=storage_backend,
        ):
            named = vol_spec.get("name", vol_name)
            vol_spec.update(_swarm_nfs_driver_opts(dir_var_lib, str(named)))
            if isinstance(nfs_meta, dict):
                vol_spec["x-infinito-nfs"] = dict(nfs_meta)

    payload: dict[str, Any] = {"volumes": to_plain(volumes)}
    if configs:
        payload["configs"] = to_plain(configs)
    if secrets:
        payload["secrets"] = to_plain(secrets)

    return dump_yaml_str(payload).rstrip()


class LookupModule(LookupBase):
    def _with_database_realign(
        self,
        applications: dict[str, Any],
        application_id: str,
        vars_: dict[str, Any],
        extra_configs: dict[str, Any] | None,
    ) -> dict[str, Any] | None:
        """Add the realignment statements an app's own mariadb reads at start.

        Args:
            applications: the merged applications mapping.
            application_id: the app whose stack is being rendered.
            vars_: the variables in scope for the lookup.
            extra_configs: configs a caller already supplied.

        The file lands as a config rather than a bind mount because a swarm task
        may run on a node the source path does not exist on.

        A role declaring two engines is left to ``compose_volumes``, which turns
        the same ValueError into the message that says which one to drop.
        """
        try:
            database_service_key = resolve_database_service_key(
                applications, application_id
            )
        except ValueError:
            return extra_configs
        if database_service_key != "mariadb":
            return extra_configs
        if get_database_service_config(applications, application_id).get("shared"):
            return extra_configs

        source = lookup_loader.get(
            "database", loader=self._loader, templar=getattr(self, "_templar", None)
        ).run([application_id, "realign_sql"], variables=vars_)[0]
        if not source:
            return extra_configs

        entity = get_entity_name(application_id)
        return {
            **(extra_configs or {}),
            REALIGN_CONFIG_KEY: {
                "name": _config_secret_name(entity, REALIGN_CONFIG_KEY, source),
                "file": source,
            },
        }

    def run(
        self,
        terms: list[Any] | None,
        variables: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        if terms:
            raise AnsibleError(
                "compose_volumes lookup expects no terms; application_id "
                "is read from the templating context"
            )

        vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
        templar = getattr(self, "_templar", None)

        application_id = vars_.get("application_id")
        if templar is not None and application_id is not None:
            application_id = templar.template(application_id)
        application_id = str(application_id or "").strip()
        if not application_id:
            raise AnsibleError(
                "compose_volumes lookup: 'application_id' is not set in the "
                "templating context"
            )

        applications = lookup_loader.get(
            "applications", loader=self._loader, templar=getattr(self, "_templar", None)
        ).run([], variables=vars_)[0]

        deployment_mode = kwargs.get("deployment_mode")
        if deployment_mode is None:
            mode_force = vars_.get("compose_mode_force", "")
            if templar is not None:
                with contextlib.suppress(Exception):
                    mode_force = templar.template(mode_force)
            deployment_mode = mode_force or vars_.get("DEPLOYMENT_MODE", "compose")

        storage = kwargs.get("storage")
        if storage is None:
            storage = vars_.get("storage")

        dir_var_lib = kwargs.get("dir_var_lib")
        if dir_var_lib is None:
            dir_var_lib = vars_["DIR_VAR_LIB"]

        if templar is not None:
            with contextlib.suppress(Exception):
                deployment_mode = templar.template(deployment_mode)
            with contextlib.suppress(Exception):
                storage = templar.template(storage)
            with contextlib.suppress(Exception):
                dir_var_lib = templar.template(dir_var_lib)

        render_jinja = None
        if templar is not None:

            def render_jinja(expr: str) -> Any:
                if not isinstance(expr, str):
                    return expr
                return templar.template(
                    _trust_as_template(expr),
                    fail_on_undefined=False,
                )

        rendered = compose_volumes(
            applications,
            application_id,
            extra_volumes=kwargs.get("extra_volumes"),
            extra_configs=self._with_database_realign(
                applications, application_id, vars_, kwargs.get("extra_configs")
            ),
            extra_secrets=kwargs.get("extra_secrets"),
            deployment_mode=str(deployment_mode).strip(),
            storage=storage,
            dir_var_lib=str(dir_var_lib).strip(),
            render_jinja=render_jinja,
        )
        return [rendered]