"""Docker lifecycle and seeding helpers for the live-Gitea integration tests.

Everything here talks to the ``docker`` CLI through ``subprocess`` (no extra
Python dependency). The harness never builds images, never uses host bind
mounts and never prunes anything: it only removes the container and the
network it created itself, identified by a unique per-run id.

Secrets (admin password, tokens) are random per run, only ever live in memory,
and are excluded from ``repr`` and from every error message.
"""

from __future__ import annotations

import atexit
import contextlib
import os
import re
import secrets
import signal
import socket
import subprocess
import sys
import threading
import time
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from types import FrameType
from typing import Any

import httpx

# Single source of truth for the Gitea image (digest-pinned).
GITEA_IMAGE = (
    "gitea/gitea:28.1.0@sha256:f505a0d3a41323b786e05942fc7a4075e07ae7c169fd9485649aa55ee543a21a"
)
NAME_PREFIX = "aegis-it-"
GITEA_PORT = 3000
ADMIN_LOGIN = "aegis-admin"
ADMIN_EMAIL = "aegis-admin@example.test"
ADMIN_PASSWORD_ENV = "GITEA_ADMIN_PASSWORD"
READY_TIMEOUT_SECONDS = 90.0
NETWORK_MODE_ENV = "AEGIS_IT_NETWORK_MODE"
HTTP_TIMEOUT_SECONDS = 30.0


class HarnessError(RuntimeError):
    """Raised for harness failures. Messages never contain secrets."""


# ---------------------------------------------------------------------------
# docker CLI helpers
# ---------------------------------------------------------------------------


def run_docker(
    *args: str,
    check: bool = True,
    timeout: float = 120.0,
    env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    """Run ``docker <args>`` and return the completed process.

    Secrets must not go into ``args`` (argv is visible to every local user): pass them
    in ``env`` and forward them with ``docker exec -e NAME``. On failure the error names
    only the first two arguments plus the command's stderr.
    """
    try:
        proc = subprocess.run(
            ["docker", *args],
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,
            env={**os.environ, **env} if env else None,
        )
    except (FileNotFoundError, PermissionError) as exc:
        raise HarnessError("the 'docker' CLI was not found or is not executable") from exc
    except subprocess.TimeoutExpired as exc:
        raise HarnessError(f"docker {' '.join(args[:2])} timed out after {timeout}s") from exc
    if check and proc.returncode != 0:
        raise HarnessError(
            f"docker {' '.join(args[:2])} failed (exit {proc.returncode}): {proc.stderr.strip()}"
        )
    return proc


_MOUNTINFO_ID_RE = re.compile(r"/containers/([0-9a-f]{64})/")
_CGROUP_ID_RE = re.compile(r"(?<![0-9a-f])([0-9a-f]{64})(?![0-9a-f])")


def container_id_candidates(
    environ: Mapping[str, str] | None = None,
    proc_root: Path = Path("/"),
) -> list[str]:
    """Return references that may name the container this process runs in.

    Order: ``$HOSTNAME``, ``/etc/hostname``, then 64-hex container ids found in
    ``/proc/self/mountinfo`` and ``/proc/self/cgroup``. On act_runner the hostname
    is normally the container id, but a runner may set another hostname or use host
    networking settings, so the other sources act as fallbacks.
    """
    env = os.environ if environ is None else environ
    candidates: list[str] = []

    def add(value: str) -> None:
        value = value.strip()
        if value and value not in candidates:
            candidates.append(value)

    add(env.get("HOSTNAME", ""))
    for rel in ("etc/hostname",):
        with contextlib.suppress(OSError):
            add((proc_root / rel).read_text(encoding="utf-8"))
    for rel, pattern in (
        ("proc/self/mountinfo", _MOUNTINFO_ID_RE),
        ("proc/self/cgroup", _CGROUP_ID_RE),
    ):
        with contextlib.suppress(OSError):
            for match in pattern.finditer((proc_root / rel).read_text(encoding="utf-8")):
                add(match.group(1))
    return candidates


def job_container_ref() -> str:
    """Return a reference docker accepts for the container running the tests.

    Each candidate from ``container_id_candidates`` is checked with
    ``docker inspect``; the first one docker resolves wins.
    """
    for candidate in container_id_candidates():
        proc = run_docker(
            "container", "inspect", "--format", "{{.Id}}", candidate, check=False, timeout=30
        )
        if proc.returncode == 0 and proc.stdout.strip():
            return candidate
    raise HarnessError(
        "AEGIS_IT_NETWORK_MODE=container: could not identify the container running the "
        "tests ($HOSTNAME, /etc/hostname and /proc/self did not resolve to a container)"
    )


def require_docker() -> None:
    """Fail with a clear message when the docker daemon is not usable."""
    try:
        proc = run_docker("version", "--format", "{{.Server.Version}}", check=False, timeout=30)
    except HarnessError as exc:
        raise HarnessError(
            f"AEGIS_INTEGRATION=1 but Docker is unavailable: {exc}. "
            "Install/start Docker or unset AEGIS_INTEGRATION."
        ) from exc
    if proc.returncode != 0:
        raise HarnessError(
            "AEGIS_INTEGRATION=1 but the Docker daemon is not reachable "
            f"({proc.stderr.strip() or 'no output'}). Start Docker or unset AEGIS_INTEGRATION."
        )


def ensure_image(image: str = GITEA_IMAGE) -> None:
    """Pull the pinned image if it is not present locally."""
    if run_docker("image", "inspect", image, check=False).returncode == 0:
        return
    run_docker("pull", image, timeout=600)


# ---------------------------------------------------------------------------
# cleanup (idempotent, registered with atexit and signal handlers)
# ---------------------------------------------------------------------------


@dataclass
class RunResources:
    """Docker resources owned by one harness run."""

    run_id: str
    container: str
    network: str
    attached_container: str | None = None  # job container joined in CI mode
    cleaned: bool = False


_ACTIVE: dict[str, RunResources] = {}
_LOCK = threading.Lock()
_HOOKS_INSTALLED = False
_PREVIOUS_HANDLERS: dict[int, Any] = {}


def cleanup_resources(res: RunResources) -> None:
    """Remove this run's container and network (idempotent, best effort)."""
    with _LOCK:
        if res.cleaned:
            return
        res.cleaned = True
        _ACTIVE.pop(res.run_id, None)
    if res.attached_container:
        run_docker("network", "disconnect", "-f", res.network, res.attached_container, check=False)
    run_docker("rm", "-f", "-v", res.container, check=False)
    run_docker("network", "rm", res.network, check=False)


def cleanup_all() -> None:
    """Clean every run still registered (atexit and signal path)."""
    for res in list(_ACTIVE.values()):
        # Best effort during interpreter exit: never let cleanup raise.
        with contextlib.suppress(Exception):
            cleanup_resources(res)


def _signal_handler(signum: int, frame: FrameType | None) -> None:
    cleanup_all()
    previous = _PREVIOUS_HANDLERS.get(signum, signal.SIG_DFL)
    signal.signal(signum, previous if previous is not None else signal.SIG_DFL)
    if callable(previous):
        previous(signum, frame)
    else:
        os.kill(os.getpid(), signum)


def _install_exit_hooks() -> None:
    global _HOOKS_INSTALLED
    if _HOOKS_INSTALLED:
        return
    _HOOKS_INSTALLED = True
    atexit.register(cleanup_all)
    if threading.current_thread() is not threading.main_thread():
        return
    for name in ("SIGTERM", "SIGHUP"):
        signum = getattr(signal, name, None)
        if signum is None:
            continue
        _PREVIOUS_HANDLERS[signum] = signal.getsignal(signum)
        signal.signal(signum, _signal_handler)


def register_resources(res: RunResources) -> None:
    """Track a run so that atexit/signal handlers can clean it up."""
    _install_exit_hooks()
    with _LOCK:
        _ACTIVE[res.run_id] = res


# ---------------------------------------------------------------------------
# Gitea instance
# ---------------------------------------------------------------------------


def unique(prefix: str) -> str:
    """Return ``<prefix>-<8 hex>``, valid as a Gitea user, org or repo name."""
    return f"{prefix}-{secrets.token_hex(4)}"


def _free_loopback_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def _gitea_env(root_url: str) -> dict[str, str]:
    """Return the ``GITEA__section__KEY`` environment for the throwaway instance."""
    return {
        "GITEA__database__DB_TYPE": "sqlite3",
        "GITEA__security__INSTALL_LOCK": "true",
        "GITEA__security__SECRET_KEY": secrets.token_hex(32),
        "GITEA__security__INTERNAL_TOKEN": secrets.token_hex(32),
        "GITEA__server__ROOT_URL": root_url,
        "GITEA__server__HTTP_PORT": str(GITEA_PORT),
        "GITEA__server__DISABLE_SSH": "true",
        "GITEA__server__OFFLINE_MODE": "true",
        "GITEA__picture__DISABLE_GRAVATAR": "true",
        "GITEA__oauth2__ENABLED": "true",
        "GITEA__oauth2__JWT_SECRET": secrets.token_urlsafe(32),
        "GITEA__actions__ENABLED": "true",
        "GITEA__service__DISABLE_REGISTRATION": "true",
        "GITEA__repository__ENABLE_PUSH_CREATE_USER": "false",
        "GITEA__repository__ENABLE_PUSH_CREATE_ORG": "false",
        "GITEA__mailer__ENABLED": "false",
        "GITEA__webhook__ALLOWED_HOST_LIST": "*",
        "GITEA__packages__ENABLED": "true",
        "GITEA__api__MAX_RESPONSE_ITEMS": "50",
    }


def _failure(response: httpx.Response, action: str) -> HarnessError:
    return HarnessError(f"{action} failed: HTTP {response.status_code}: {response.text[:300]}")


@dataclass
class GiteaInstance:
    """A running throwaway Gitea plus helpers to seed and call it."""

    base_url: str
    admin_login: str
    admin_password: str = field(repr=False)
    admin_token: str = field(repr=False)
    container: str
    network: str
    run_id: str
    _client: httpx.Client = field(repr=False, default_factory=httpx.Client)

    def api(
        self,
        method: str,
        path: str,
        token: str | None = None,
        *,
        anonymous: bool = False,
        **kwargs: Any,
    ) -> httpx.Response:
        """Call ``/api/v1<path>``.

        ``token`` defaults to the admin token; pass ``anonymous=True`` for an
        unauthenticated call, or ``auth=(login, password)`` for basic auth.
        """
        headers: dict[str, str] = dict(kwargs.pop("headers", None) or {})
        if not anonymous and "auth" not in kwargs:
            headers["Authorization"] = f"token {token or self.admin_token}"
        url = f"{self.base_url}/api/v1/{path.lstrip('/')}"
        return self._client.request(method, url, headers=headers, **kwargs)

    def unique(self, prefix: str) -> str:
        """Return ``<prefix>-<8 hex>``, valid as a Gitea user, org or repo name."""
        return unique(prefix)

    def create_user(self, login: str, admin: bool = False) -> tuple[str, str]:
        """Create a user with a random password and return ``(login, password)``."""
        password = secrets.token_urlsafe(24)
        response = self.api(
            "POST",
            "/admin/users",
            json={
                "username": login,
                "email": f"{login}@example.test",
                "password": password,
                "must_change_password": False,
            },
        )
        if response.status_code >= 300:
            raise _failure(response, f"create user {login}")
        if admin:
            promote = self.api(
                "PATCH",
                f"/admin/users/{login}",
                json={"login_name": login, "source_id": 0, "admin": True},
            )
            if promote.status_code >= 300:
                raise _failure(promote, f"promote user {login}")
        return login, password

    def mint_token(self, login: str, password: str, scopes: list[str]) -> str:
        """Create an access token for ``login`` (basic auth) and return its secret."""
        response = self.api(
            "POST",
            f"/users/{login}/tokens",
            auth=(login, password),
            json={"name": f"it-{secrets.token_hex(4)}", "scopes": scopes},
        )
        if response.status_code >= 300:
            raise _failure(response, f"mint token for {login}")
        token = response.json().get("sha1")
        if not isinstance(token, str) or not token:
            raise HarnessError(f"mint token for {login} returned no token")
        return token

    def create_org(self, name: str) -> dict[str, Any]:
        """Create an organization (as admin) and return its JSON."""
        response = self.api("POST", "/orgs", json={"username": name})
        if response.status_code >= 300:
            raise _failure(response, f"create org {name}")
        data: dict[str, Any] = response.json()
        return data

    def create_repo(
        self, owner: str, name: str, auto_init: bool = True, org: bool = False
    ) -> dict[str, Any]:
        """Create a repository for a user or an org (as admin) and return its JSON."""
        path = f"/orgs/{owner}/repos" if org else f"/admin/users/{owner}/repos"
        response = self.api(
            "POST",
            path,
            json={"name": name, "auto_init": auto_init, "default_branch": "main"},
        )
        if response.status_code >= 300:
            raise _failure(response, f"create repo {owner}/{name}")
        data: dict[str, Any] = response.json()
        return data

    def close(self) -> None:
        """Close the HTTP client."""
        self._client.close()


def _container_logs_tail(container: str, lines: int = 100) -> str:
    proc = run_docker("logs", "--tail", str(lines), container, check=False, timeout=30)
    return (proc.stdout + proc.stderr).strip()


def _wait_ready(base_url: str, container: str) -> None:
    deadline = time.monotonic() + READY_TIMEOUT_SECONDS
    last_error = "no response"
    while time.monotonic() < deadline:
        state = run_docker(
            "inspect", "--format", "{{.State.Running}}", container, check=False, timeout=15
        )
        if state.stdout.strip() == "false":
            last_error = "container exited"
            break
        try:
            response = httpx.get(f"{base_url}/api/v1/version", timeout=3.0)
            if response.status_code == 200:
                return
            last_error = f"HTTP {response.status_code}"
        except httpx.HTTPError as exc:
            last_error = type(exc).__name__
        time.sleep(1.0)
    print(
        f"--- gitea container logs (tail 100) ---\n{_container_logs_tail(container)}",
        file=sys.stderr,
    )
    raise HarnessError(
        f"Gitea at {base_url} was not ready within {READY_TIMEOUT_SECONDS:.0f}s ({last_error})"
    )


def _start_container(res: RunResources, container_mode: bool) -> str:
    """Create the network, start Gitea and return the base URL the tests use."""
    run_docker("network", "create", res.network)
    last_error: HarnessError | None = None
    attempts = 1 if container_mode else 3
    for _ in range(attempts):
        if container_mode:
            base_url = f"http://{res.container}:{GITEA_PORT}"
            publish: list[str] = []
        else:
            # ROOT_URL must match how tests reach Gitea, so the host port is chosen
            # up front (loopback only) instead of letting Docker pick one.
            port = _free_loopback_port()
            base_url = f"http://127.0.0.1:{port}"
            publish = ["-p", f"127.0.0.1:{port}:{GITEA_PORT}"]
        env_args: list[str] = []
        for key, value in _gitea_env(base_url).items():
            env_args += ["-e", f"{key}={value}"]
        try:
            run_docker(
                "run",
                "-d",
                "--name",
                res.container,
                "--network",
                res.network,
                "--label",
                "aegis-it=1",
                *publish,
                *env_args,
                GITEA_IMAGE,
            )
        except HarnessError as exc:
            last_error = exc
            run_docker("rm", "-f", "-v", res.container, check=False)
            continue
        if not container_mode:
            mapped = run_docker("port", res.container, f"{GITEA_PORT}/tcp").stdout
            if not any(line.strip().endswith(f":{port}") for line in mapped.splitlines()):
                raise HarnessError("published port does not match the configured ROOT_URL")
        return base_url
    raise HarnessError(f"could not start the Gitea container: {last_error}")


def start_gitea() -> GiteaInstance:
    """Start a throwaway Gitea, seed the admin user and token, return the instance.

    Raises ``HarnessError`` (after cleaning up) on any failure.
    """
    require_docker()
    ensure_image()
    run_id = secrets.token_hex(4)
    container_mode = os.environ.get(NETWORK_MODE_ENV, "").lower() == "container"
    res = RunResources(
        run_id=run_id, container=f"{NAME_PREFIX}{run_id}", network=f"{NAME_PREFIX}{run_id}"
    )
    register_resources(res)
    try:
        base_url = _start_container(res, container_mode)
        if container_mode:
            job_container = job_container_ref()
            run_docker("network", "connect", res.network, job_container)
            res.attached_container = job_container
        _wait_ready(base_url, res.container)
        admin_password = secrets.token_urlsafe(24)
        # The password travels in the environment of the docker CLI process and is forwarded
        # with ``-e NAME`` (no value), so it never appears in any argv on the host. Inside
        # the container it is read from the shell environment, not from the command line.
        run_docker(
            "exec",
            "-e",
            ADMIN_PASSWORD_ENV,
            "-u",
            "git",
            res.container,
            "sh",
            "-c",
            'exec gitea admin user create --admin --username "$1" --email "$2" '
            f'--password "${ADMIN_PASSWORD_ENV}" --must-change-password=false',
            "sh",
            ADMIN_LOGIN,
            ADMIN_EMAIL,
            env={ADMIN_PASSWORD_ENV: admin_password},
        )
        instance = GiteaInstance(
            base_url=base_url,
            admin_login=ADMIN_LOGIN,
            admin_password=admin_password,
            admin_token="",
            container=res.container,
            network=res.network,
            run_id=run_id,
            _client=httpx.Client(timeout=HTTP_TIMEOUT_SECONDS),
        )
        instance.admin_token = instance.mint_token(ADMIN_LOGIN, admin_password, ["all"])
        return instance
    except BaseException:
        cleanup_resources(res)
        raise


def stop_gitea(instance: GiteaInstance) -> None:
    """Close the client and remove the instance's Docker resources."""
    instance.close()
    with _LOCK:
        res = _ACTIVE.get(instance.run_id)
    if res is not None:
        cleanup_resources(res)