"""Docker volume management + Compose volume YAML generation."""
from __future__ import annotations

import logging
import threading
import time
from typing import Callable, Literal, Optional

import yaml

from docker_client import get_client, get_long_client, safe_call
from services.prune_job import PruneJob, PrunePlan

logger = logging.getLogger(__name__)

COMPOSE_PROJECT_LABEL = "com.docker.compose.project"
#: Set by the daemon (Engine 23+) on volumes created without a name. A plain
#: `docker volume prune` only removes these; `-a` removes named ones too.
ANONYMOUS_LABEL = "com.docker.volume.anonymous"

# `docker system df -v` walks every volume's contents, so it can take many
# seconds. Cache the result so the (polled) UI and repeated requests reuse it.
_SIZE_TTL = 60.0
_size_cache: dict = {"at": 0.0, "data": {}}
_size_lock = threading.Lock()


def volume_sizes(force: bool = False) -> dict:
    """Return {volume_name: size_bytes|None}. Cached (~60s) since it's expensive."""
    now = time.time()
    with _size_lock:
        if not force and _size_cache["data"] and now - _size_cache["at"] < _SIZE_TTL:
            return _size_cache["data"]

    # The slow call this cache exists for; the 60s client would time out on a
    # host with large volumes.
    client = get_long_client()
    df = safe_call(client.df)
    sizes: dict[str, Optional[int]] = {}
    for v in df.get("Volumes") or []:
        ud = v.get("UsageData") or {}
        size = ud.get("Size")
        sizes[v.get("Name")] = size if isinstance(size, int) and size >= 0 else None

    with _size_lock:
        _size_cache["at"] = time.time()
        _size_cache["data"] = sizes
    return sizes


# --------------------------------------------------------------------------- #
# Listing / pruning
# --------------------------------------------------------------------------- #


def list_volumes() -> list[dict]:
    client = get_client()
    volumes = safe_call(client.volumes.list)
    # Map volume name -> set of containers using it.
    containers = safe_call(client.containers.list, all=True)
    usage: dict[str, list[str]] = {}
    for c in containers:
        for mount in c.attrs.get("Mounts", []) or []:
            if mount.get("Type") == "volume" and mount.get("Name"):
                usage.setdefault(mount["Name"], []).append(c.name)

    result = []
    for v in volumes:
        attrs = v.attrs
        name = v.name
        result.append(
            {
                "name": name,
                "driver": attrs.get("Driver"),
                "mountpoint": attrs.get("Mountpoint"),
                "created_at": attrs.get("CreatedAt"),
                "labels": attrs.get("Labels") or {},
                "scope": attrs.get("Scope"),
                "stack": (attrs.get("Labels") or {}).get(COMPOSE_PROJECT_LABEL),
                "used_by": usage.get(name, []),
                "in_use": bool(usage.get(name)),
            }
        )
    return result


def orphaned_volumes() -> list[dict]:
    return [v for v in list_volumes() if not v["in_use"]]


def remove_volume(name: str, force: bool = False) -> None:
    client = get_client()
    volume = safe_call(client.volumes.get, name)
    safe_call(volume.remove, force=force)


_prune = PruneJob("volume")


def prune_status() -> dict:
    return _prune.status()


def _candidates(client, all_unused: bool) -> list[dict]:
    """The volumes ``docker volume prune [-a]`` would remove.

    Unused means no container, running or stopped, mounts it. Without -a only
    anonymous volumes qualify. Volumes created before Engine 23 carry no
    anonymous label and so are only removed with -a, which errs on the side of
    keeping data.
    """
    in_use = set()
    for c in safe_call(client.api.containers, all=True):
        for mount in c.get("Mounts") or []:
            if mount.get("Type") == "volume" and mount.get("Name"):
                in_use.add(mount["Name"])
    volumes = (safe_call(client.api.volumes) or {}).get("Volumes") or []
    return [
        v
        for v in volumes
        if v["Name"] not in in_use
        # Cluster (swarm) volumes are not the local daemon's to prune.
        and v.get("Scope", "local") == "local"
        and (all_unused or ANONYMOUS_LABEL in (v.get("Labels") or {}))
    ]


def _label(volume: dict) -> str:
    name = volume["Name"]
    # Anonymous volumes are 64 hex characters; the first 12 identify them.
    return name[:12] if ANONYMOUS_LABEL in (volume.get("Labels") or {}) else name


def _plan(all_unused: bool) -> PrunePlan:
    client = get_long_client()
    measured: dict = {}

    def prepare() -> None:
        # Volumes share no data, so the freed space is simply the sum of what
        # was deleted, provided their sizes are known up front. Unused volumes
        # cannot change, so the cached figures are good enough.
        try:
            measured["sizes"] = volume_sizes()
        except Exception as exc:  # noqa: BLE001 - the size is a nicety, not the job
            logger.warning("Could not measure volume sizes: %s", exc)

    def remove(volume: dict) -> tuple[list, bool]:
        # No force: a volume mounted since the listing makes the daemon refuse.
        safe_call(client.api.remove_volume, volume["Name"])
        return [volume["Name"]], True

    def finish(names: list) -> tuple[dict, Optional[int]]:
        # No closing daemon prune as for images: nothing is left behind, and
        # on a pre-23 daemon an unfiltered one would delete named volumes too.
        sizes = measured.get("sizes")
        reclaimed = (
            sum(sizes.get(name) or 0 for name in names) if sizes is not None else None
        )
        with _size_lock:
            _size_cache["at"] = 0.0  # deleted volumes must not linger in it
        return {"VolumesDeleted": names, "SpaceReclaimed": reclaimed or 0}, reclaimed

    return PrunePlan(
        all=all_unused,
        items=_candidates(client, all_unused),
        label=_label,
        remove=remove,
        finish=finish,
        prepare=prepare,
    )


def prune_volumes(all_unused: bool = False) -> dict:
    """Remove unused volumes and wait for the result.

    By default only anonymous volumes; ``all_unused=True`` also removes named
    ones (``docker volume prune -a``). Progress is visible through
    prune_status() while it runs.
    """
    return _prune.run(lambda: _plan(all_unused))


def start_prune(
    all_unused: bool = False, on_done: Optional[Callable[[dict], None]] = None
) -> dict:
    """Start a prune in the background and return its initial status; the UI
    polls prune_status(). ``on_done`` gets what prune_volumes() would return."""
    return _prune.start(lambda: _plan(all_unused), on_done)


# --------------------------------------------------------------------------- #
# YAML generation
# --------------------------------------------------------------------------- #

VolumeType = Literal["bind", "named", "nfs", "smb", "tmpfs"]


def _nfs_options(server: str, opts: dict) -> str:
    """Build the `o:` option string for an NFS driver_opts block."""
    parts = [f"addr={server}"]
    # rw / ro
    if opts.get("rw", True):
        parts.append("rw")
    else:
        parts.append("ro")
    for flag in ("soft", "hard", "nolock", "noatime", "noacl", "nocto", "bg"):
        if opts.get(flag):
            parts.append(flag)
    version = opts.get("nfsvers") or opts.get("version")
    if version:
        parts.append(f"nfsvers={version}")
    if opts.get("timeo") is not None:
        parts.append(f"timeo={opts['timeo']}")
    if opts.get("retrans") is not None:
        parts.append(f"retrans={opts['retrans']}")
    extra = (opts.get("extra") or "").strip()
    if extra:
        parts.append(extra.lstrip(","))
    return ",".join(parts)


def _smb_options(opts: dict) -> str:
    parts = []
    if opts.get("username"):
        parts.append(f"username={opts['username']}")
    if opts.get("password") is not None:
        parts.append(f"password={opts['password']}")
    if opts.get("uid") is not None:
        parts.append(f"uid={opts['uid']}")
    if opts.get("gid") is not None:
        parts.append(f"gid={opts['gid']}")
    version = opts.get("vers")
    if version:
        parts.append(f"vers={version}")
    if opts.get("noperm"):
        parts.append("noperm")
    if opts.get("file_mode"):
        parts.append(f"file_mode={opts['file_mode']}")
    if opts.get("dir_mode"):
        parts.append(f"dir_mode={opts['dir_mode']}")
    extra = (opts.get("extra") or "").strip()
    if extra:
        parts.append(extra.lstrip(","))
    return ",".join(parts)


def build_volume_definition(spec: dict) -> dict:
    """Return the top-level `volumes:` declaration for a named/nfs/smb volume.

    Returns {} for bind/tmpfs (those live entirely on the service).
    """
    vtype: VolumeType = spec["type"]
    name = spec.get("volume_name") or spec.get("name")

    if vtype == "named":
        decl: dict = {"driver": spec.get("driver", "local")}
        if spec.get("driver_opts"):
            decl["driver_opts"] = spec["driver_opts"]
        if not decl.get("driver_opts") and decl["driver"] == "local":
            # A plain named volume can be declared as null.
            return {name: None}
        return {name: decl}

    if vtype == "nfs":
        o = _nfs_options(spec["nfs_server"], spec.get("options", {}))
        device = spec["nfs_path"]
        if not device.startswith(":"):
            device = ":" + device
        return {
            name: {
                "driver": "local",
                "driver_opts": {"type": "nfs", "o": o, "device": device},
            }
        }

    if vtype == "smb":
        o = _smb_options(spec.get("options", {}))
        device = spec["smb_share"]  # //server/share
        return {
            name: {
                "driver": "local",
                "driver_opts": {"type": "cifs", "o": o, "device": device},
            }
        }

    return {}


def build_service_mount(spec: dict) -> str | dict:
    """Return the entry to add to a service's `volumes:` list."""
    vtype: VolumeType = spec["type"]
    if vtype == "bind":
        mode = "ro" if not spec.get("options", {}).get("rw", True) else "rw"
        return f"{spec['host_path']}:{spec['container_path']}:{mode}"
    if vtype == "tmpfs":
        return {
            "type": "tmpfs",
            "target": spec["container_path"],
            "tmpfs": {
                k: v
                for k, v in {
                    "size": spec.get("options", {}).get("size"),
                    "mode": spec.get("options", {}).get("mode"),
                }.items()
                if v is not None
            },
        }
    # named / nfs / smb
    name = spec.get("volume_name") or spec.get("name")
    return f"{name}:{spec['container_path']}"


def generate_yaml(spec: dict) -> str:
    """Produce a ready-to-insert YAML fragment for the given volume spec.

    Includes the top-level `volumes:` block (if any) and a `services:` example
    showing the mount, mirroring the wizard preview.
    """
    doc: dict = {}
    definition = build_volume_definition(spec)
    if definition:
        doc["volumes"] = definition
    service_name = spec.get("service", "service")
    doc["services"] = {service_name: {"volumes": [build_service_mount(spec)]}}
    return yaml.safe_dump(doc, sort_keys=False, default_flow_style=False)