headquarter/apps/api/src/services/docker/tunnel.py

"""Cloudflare tunnel management using cloudflared Docker containers.

Each tunnel runs as a Docker container on the same 'backend' network as the API.
cloudflared connects to the tool container by its Docker Compose service name
(e.g. http://code-server-headquarter-34837cd3:8443).
"""

import logging
import re
import subprocess
from typing import Any

from src.services.docker.container import get_backend_network_name

logger = logging.getLogger(__name__)

TUNNEL_IMAGE = "cloudflare/cloudflared:latest"


def _tunnel_container_name(instance_name: str) -> str:
    return f"tunnel-{instance_name.lower()}"


def _ensure_image() -> None:
    """Pull cloudflared image if not already present."""
    result = subprocess.run(
        ["docker", "images", "-q", TUNNEL_IMAGE],
        capture_output=True,
        text=True,
    )
    if not result.stdout.strip():
        logger.info("Pulling %s ...", TUNNEL_IMAGE)
        pull = subprocess.run(
            ["docker", "pull", TUNNEL_IMAGE],
            capture_output=True,
            text=True,
        )
        if pull.returncode != 0:
            logger.warning("Failed to pull %s: %s", TUNNEL_IMAGE, pull.stderr)


def _cleanup_stale_tunnel(tunnel_name: str) -> None:
    """Remove any existing tunnel container with this name."""
    subprocess.run(
        ["docker", "stop", "-t", "3", tunnel_name],
        capture_output=True,
        text=True,
    )
    subprocess.run(
        ["docker", "rm", "-f", tunnel_name],
        capture_output=True,
        text=True,
    )


def _get_tunnel_logs(tunnel_name: str) -> tuple[str, str]:
    """Get stdout and stderr logs from a container."""
    result = subprocess.run(
        ["docker", "logs", tunnel_name],
        capture_output=True,
        text=True,
    )
    return result.stdout, result.stderr


def _get_tunnel_exit_code(tunnel_name: str) -> int | None:
    """Get exit code of a container if it has exited."""
    result = subprocess.run(
        ["docker", "inspect", "-f", "{{.State.ExitCode}}", tunnel_name],
        capture_output=True,
        text=True,
    )
    if result.returncode == 0:
        try:
            return int(result.stdout.strip())
        except ValueError:
            pass
    return None


def start_tunnel(
    instance_name: str,
    container_port: int,
    timeout: int = 30,
    target_url: str | None = None,
) -> dict[str, str]:
    """Start a temporary Cloudflare tunnel for an instance.

    Args:
        instance_name: The tool instance name (used for tunnel naming).
        container_port: The port the tool container listens on internally.
        timeout: Seconds to wait for the tunnel URL.
        target_url: Optional explicit URL to proxy to. If omitted, derives
            http://{instance_name.lower()}:{container_port}.

    Returns:
        Dict with 'url' and 'container_name'.
    """
    _ensure_image()

    tunnel_name = _tunnel_container_name(instance_name)
    _cleanup_stale_tunnel(tunnel_name)

    # Target the tool container by name on the backend network
    if target_url is None:
        target_url = f"http://{instance_name.lower()}:{container_port}"

    cmd = [
        "docker",
        "run",
        "-d",
        "--network",
        get_backend_network_name(),
        "--name",
        tunnel_name,
        TUNNEL_IMAGE,
        "tunnel",
        "--no-autoupdate",
        "--url",
        target_url,
    ]

    logger.debug("Running: %s", " ".join(cmd))
    proc = subprocess.run(cmd, capture_output=True, text=True)
    if proc.returncode != 0:
        raise RuntimeError(
            f"Failed to start tunnel container {tunnel_name}: {proc.stderr}"
        )

    container_id = proc.stdout.strip()
    logger.debug("Tunnel container started: %s", container_id)

    # Wait for URL to appear in logs
    # Exclude api.trycloudflare.com which is the Cloudflare API endpoint,
    # not a tunnel URL. Real tunnel URLs have random subdomains (10+ chars).
    url_pattern = re.compile(r"https://(?!api\.)[a-z0-9-]{10,}\.trycloudflare\.com")
    start_time = __import__("time").time()
    url: str | None = None
    combined_logs = ""

    while __import__("time").time() - start_time < timeout:
        stdout, stderr = _get_tunnel_logs(tunnel_name)
        combined_logs = stdout + "\n" + stderr

        match = url_pattern.search(combined_logs)
        if match:
            url = match.group(0)
            break

        # Check if container exited early
        exit_code = _get_tunnel_exit_code(tunnel_name)
        if exit_code is not None and exit_code != 0:
            _cleanup_stale_tunnel(tunnel_name)
            raise RuntimeError(
                f"Tunnel container {tunnel_name} exited with code {exit_code}. "
                f"Logs:\n{combined_logs[-3000:]}"
            )

        __import__("time").sleep(0.5)

    if not url:
        stdout, stderr = _get_tunnel_logs(tunnel_name)
        combined_logs = stdout + "\n" + stderr
        exit_code = _get_tunnel_exit_code(tunnel_name)

        _cleanup_stale_tunnel(tunnel_name)
        raise RuntimeError(
            f"Tunnel {tunnel_name} did not produce a URL within {timeout}s. "
            f"Exit code: {exit_code}. Logs:\n{combined_logs[-3000:]}"
        )

    # Wait a moment for Cloudflare DNS edge to propagate the new tunnel subdomain
    __import__("time").sleep(2)

    logger.info(
        "Tunnel %s started for %s → %s (%s)",
        tunnel_name,
        instance_name,
        target_url,
        url,
    )
    return {"url": url, "container_name": tunnel_name}


def stop_tunnel(instance_name: str) -> None:
    """Stop and remove the tunnel container for an instance."""
    tunnel_name = _tunnel_container_name(instance_name)
    _cleanup_stale_tunnel(tunnel_name)
    logger.debug("Stopped and removed tunnel container %s", tunnel_name)


def recreate_tunnel(
    instance_name: str, container_port: int, target_url: str | None = None
) -> dict[str, str]:
    """Recreate a tunnel for an instance.

    Args:
        instance_name: The tool instance name.
        container_port: The port the tool container listens on internally.
        target_url: Optional explicit origin URL. If omitted, derives
            http://{instance_name.lower()}:{container_port}.
    """
    stop_tunnel(instance_name)
    return start_tunnel(instance_name, container_port, target_url=target_url)


def check_tunnel_health(url: str, timeout: int = 10) -> dict[str, Any]:
    """Check if a tunnel URL is healthy.

    Returns:
        Dict with 'tunnel_status', 'status_code', 'healthy', 'error'.
    """
    try:
        result = subprocess.run(
            [
                "curl",
                "-s",
                "-o",
                "/dev/null",
                "-w",
                "%{http_code}",
                "--max-time",
                str(timeout),
                url,
            ],
            capture_output=True,
            text=True,
            timeout=timeout + 5,
        )
        status_code = int(result.stdout.strip())

        if 200 <= status_code < 400:
            return {
                "tunnel_status": "healthy",
                "status_code": status_code,
                "healthy": True,
                "error": None,
            }
        if status_code in (502, 503, 504):
            return {
                "tunnel_status": "error_response",
                "status_code": status_code,
                "healthy": False,
                "error": f"Application returned HTTP {status_code}",
            }
        return {
            "tunnel_status": "error_response",
            "status_code": status_code,
            "healthy": False,
            "error": f"HTTP {status_code}",
        }
    except subprocess.TimeoutExpired:
        return {
            "tunnel_status": "unreachable",
            "status_code": None,
            "healthy": False,
            "error": "Tunnel request timed out",
        }
    except (ValueError, Exception) as exc:
        error_str = str(exc).lower()
        if any(
            err in error_str
            for err in [
                "connection refused",
                "econnrefused",
                "could not resolve",
                "nodename",
            ]
        ):
            return {
                "tunnel_status": "unreachable",
                "status_code": None,
                "healthy": False,
                "error": f"Tunnel unreachable: {exc}",
            }
        return {
            "tunnel_status": "unreachable",
            "status_code": None,
            "healthy": False,
            "error": str(exc),
        }