ADR 021: Proxy Plugin System for Reverse Proxy Configuration¶

class Proxy(Protocol):
    """Protocol for reverse proxy configuration strategies.

    A proxy is responsible for configuring the reverse proxy server
    (Nginx, Caddy, Traefik, etc.) to route HTTP(S) traffic to
    deployed applications.
    """

    app: App              # Application being configured
    env: Env              # Application environment variables
    workers: dict[str, str]  # Worker name -> socket path mapping

    def setup(self) -> None:
        """Configure the proxy for this application.

        Implementations must:
        1. Generate proxy configuration files
        2. Configure SSL/TLS certificates (ACME/Let's Encrypt)
        3. Set up static file serving if needed
        4. Reload/restart the proxy server

        Side effects:
        - Writes configuration files to proxy config directory
        - May request SSL certificates from ACME provider
        - Reloads proxy server (graceful reload, no downtime)

        Raises:
            RuntimeError: If proxy configuration fails
            FileNotFoundError: If required files missing
        """

{
    "web": "/tmp/myapp.sock",        # Web worker socket
    "api": "/tmp/myapp-api.sock",    # API worker socket
}

from hop3.core.hooks import hookimpl

class NginxProxyPlugin:
    """Nginx reverse proxy plugin for Hop3."""

    name = "nginx"

    @hookimpl
    def get_proxies(self) -> list:
        """Return list of proxy strategy classes."""
        return [NginxVirtualHost]

# Auto-register when module imported
plugin = NginxProxyPlugin()

@dataclass(frozen=True)
class NginxVirtualHost:
    """Nginx reverse proxy implementation."""

    app: App
    env: Env
    workers: dict[str, str]

    def setup(self) -> None:
        """Configure Nginx for this application."""
        server_name = self.env["HOST_NAME"]

        # 1. Generate upstream backends configuration
        upstream_config = self._generate_upstream(self.workers)

        # 2. Configure SSL/TLS certificates
        cert_path = self._setup_acme_certificates(server_name)

        # 3. Generate main server block
        server_config = self._generate_server_block(
            server_name=server_name,
            cert_path=cert_path,
            upstream="app_upstream",
            static_paths=self.env.get("NGINX_STATIC_PATHS", ""),
        )

        # 4. Write configuration file
        config_file = HOP3_ROOT / "nginx" / f"{self.app.name}.conf"
        config_file.write_text(upstream_config + server_config)

        # 5. Validate and reload Nginx
        run(["nginx", "-t"])  # Validate config
        run(["nginx", "-s", "reload"])  # Graceful reload

    def _generate_upstream(self, workers: dict[str, str]) -> str:
        """Generate upstream block for uWSGI sockets."""
        return f"""
upstream app_upstream {{
    server unix:{workers['web']};
}}
"""

    def _setup_acme_certificates(self, domains: str) -> Path:
        """Request/renew SSL certificates via ACME."""
        # Implementation uses acme-tiny or similar
        return Path("/etc/letsencrypt/live/example.com/fullchain.pem")

    def _generate_server_block(self, **kwargs) -> str:
        """Generate Nginx server block from template."""
        # Template rendering logic
        return """
server {
    listen 80;
    listen [::]:80;
    server_name example.com;

    location / {
        include uwsgi_params;
        uwsgi_pass app_upstream;
    }
}
"""

def get_proxy_strategy(app: App, env: Env, workers: dict[str, str]) -> Proxy:
    """Find and instantiate the appropriate proxy strategy.

    The proxy type is determined by HOP3_PROXY_TYPE environment variable.

    Args:
        app: Application to configure
        env: Application environment
        workers: Worker name -> socket path mapping

    Returns:
        Proxy instance ready to call setup()

    Raises:
        RuntimeError: If configured proxy type not found
    """
    from hop3.config import HOP3_PROXY_TYPE

    pm = get_plugin_manager()
    strategies = pm.hook.get_proxies()

    for strategy_class in flatten(strategies):
        if HOP3_PROXY_TYPE in strategy_class.__name__.lower():
            return strategy_class(app, env, workers)

    raise RuntimeError(f"Proxy type '{HOP3_PROXY_TYPE}' not found")

# Usage in deployment
proxy = get_proxy_strategy(app, env, workers)
proxy.setup()