flutter-artifacts-proxy/src/charmed_server.py

# Copyright 2026 Valentin Haudiquet
# See LICENSE file for licensing details.

"""Functions for managing and interacting with the workload.

The intention is that this module could be used outside the context of a charm.
"""

import logging
import os
import shutil
import signal
import subprocess
from pathlib import Path

from charmlibs import apt, pathops

logger = logging.getLogger(__name__)

# Get the charm directory dynamically
CHARM_DIR = Path(__file__).parent.parent
STATIC_SOURCE_DIR = CHARM_DIR / "src" / "static"

CONFIG_FILE = pathops.LocalPath("/etc/nginx/nginx.conf")
PID_FILE = pathops.LocalPath(CHARM_DIR / "nginx.pid")


def ensure_config() -> bool:
    """Ensure that nginx is configured. Return True if any changes were made."""
    config = f"""\
worker_processes  1;
pid {PID_FILE};
events {{
    worker_connections  1024;
}}
http {{
    default_type  application/octet-stream;
    server {{
        listen       80;
        root '{STATIC_SOURCE_DIR}';
        location / {{
            autoindex on;
            try_files $uri $uri/ @external;
        }}
        location @external {{
            return 301 https://storage.googleapis.com$request_uri ;
        }}
    }}
}}
"""
    return pathops.ensure_contents(CONFIG_FILE, config)


def install() -> None:
    """Install nginx apt package."""
    apt.update()
    apt.add_package("nginx", "1.18.0-6ubuntu14.8")
    subprocess.check_output(["service", "nginx", "stop"])
    os.remove("/etc/nginx/nginx.conf")
    os.unlink("/etc/nginx/sites-enabled/default")


def is_installed() -> bool:
    """Check if nginx is installed."""
    return shutil.which("nginx") is not None


def start() -> None:
    """Start the workload (by running the command)."""
    # Log static directory contents
    if STATIC_SOURCE_DIR.exists():
        files = list(STATIC_SOURCE_DIR.rglob("*"))
        logger.info(f"Serving following files in static directory: {len(files)}")
        for file in files:
            if file.is_file():
                logger.info(f"  - {file.relative_to(STATIC_SOURCE_DIR)}")
    else:
        logger.warning(f"Static directory does not exist: {STATIC_SOURCE_DIR}")

    # Check if nginx is already running
    if is_running():
        logger.info("nginx is already running")
        return

    # Ensure PID file directory exists
    pid_dir = Path(PID_FILE).parent
    pid_dir.mkdir(parents=True, exist_ok=True)

    # Touch the PID file to ensure it exists
    Path(PID_FILE).touch()

    # Test configuration first
    result = subprocess.run(["nginx", "-t"], capture_output=True, text=True)
    if result.returncode != 0:
        logger.error(f"nginx configuration test failed: {result.stderr}")
        raise RuntimeError(f"nginx configuration test failed: {result.stderr}")

    # Start nginx
    result = subprocess.run(["nginx"], capture_output=True, text=True)
    if result.returncode != 0:
        logger.error(f"Failed to start nginx: {result.stderr}")
        raise RuntimeError(f"Failed to start nginx: {result.stderr}")


def stop() -> None:
    """Stop the workload."""
    pid = _get_pid()
    if pid:
        os.kill(pid, signal.SIGTERM)


def reload_config() -> None:
    """Reload nginx configuration."""
    result = subprocess.run(["nginx", "-s", "reload"], capture_output=True, text=True)
    if result.returncode != 0:
        logger.warning(f"Failed to reload nginx config: {result.stderr}")
    else:
        logger.info("nginx configuration reloaded successfully")


def is_running() -> bool:
    """Return whether nginx is running."""
    # First try to get PID from file
    pid = _get_pid()
    if pid is not None:
        return True

    # Fallback: check if nginx process is running using pgrep
    result = subprocess.run(["pgrep", "-f", "nginx"], capture_output=True, text=True)
    return result.returncode == 0


def _get_pid() -> int | None:
    """Return the PID of the nginx process, or None if the process can't be found."""
    if not PID_FILE.exists():
        return None

    try:
        pid = int(PID_FILE.read_text())
    except (ValueError, IOError) as e:
        logger.error(f"Failed to read PID file: {e}")
        return None

    try:
        # Sending signal 0 doesn't terminate the process. It just checks whether the PID exists.
        os.kill(pid, 0)
    except ProcessLookupError:
        return None

    return pid


def get_version() -> str | None:
    """Get the running version of nginx."""
    result = subprocess.run(["nginx", "-v"], check=True, capture_output=True, text=True)
    return result.stdout.removeprefix("nginx version:").strip()