1import os
  2import subprocess
  3import sys
  4import time
  5import tomllib
  6from pathlib import Path
  7from typing import Any
  8
  9import click
 10from plain.runtime import APP_PATH, PLAIN_TEMP_PATH
 11
 12from .process import Supervisor
 13from .utils import has_pyproject_toml
 14
 15
 16def auto_start_services() -> None:
 17    """Start dev *services* in the background if not already running."""
 18
 19    # Check if we're in CI and auto-start is not explicitly enabled
 20    if os.environ.get("CI") and os.environ.get("DEV_SERVICES_AUTO") is None:
 21        return
 22
 23    if os.environ.get("DEV_SERVICES_AUTO", "true") not in [
 24        "1",
 25        "true",
 26        "yes",
 27    ]:
 28        return
 29
 30    # Only auto-start services for commands that need the database/runtime
 31    service_commands = {
 32        "postgres",
 33        "dev",
 34        "migrations",
 35        "preflight",
 36        "request",
 37        "run",
 38        "shell",
 39        "test",
 40    }
 41    if not (service_commands & set(sys.argv)):
 42        return
 43
 44    # Don't do anything if it looks like a "services" command is being run explicitly
 45    if "dev" in sys.argv and (
 46        "logs" in sys.argv or "services" in sys.argv or "--stop" in sys.argv
 47    ):
 48        return
 49
 50    if not ServicesSupervisor.get_services(APP_PATH.parent):
 51        return
 52
 53    # Cheap pre-check to avoid a needless spawn; the spawned supervisor's
 54    # acquire() is the real guarantee against double-starting.
 55    if ServicesSupervisor.running_pid():
 56        return
 57
 58    click.secho(
 59        "Starting background dev services (terminate with `plain dev --stop`)...",
 60        dim=True,
 61    )
 62
 63    ServicesSupervisor.spawn_background()
 64
 65    # Give services time to start and retry the check
 66    wait_times = [0.5, 1, 1]  # First check at 0.5s, then 1s intervals
 67    for wait_time in wait_times:
 68        time.sleep(wait_time)
 69        if ServicesSupervisor.running_pid():
 70            return  # Services started successfully
 71
 72    # Only show error after multiple attempts
 73    if not ServicesSupervisor.running_pid():
 74        click.secho(
 75            "Failed to start dev services. Here are the logs:",
 76            fg="red",
 77        )
 78        subprocess.run(
 79            ["plain", "dev", "logs", "--services"],
 80            check=False,
 81        )
 82        sys.exit(1)
 83
 84
 85class ServicesSupervisor(Supervisor):
 86    state_filename = "services.pid"
 87    log_dir = PLAIN_TEMP_PATH / "dev" / "logs" / "services"
 88    background_command = ("dev", "services")
 89    display_name = "Services"
 90
 91    @staticmethod
 92    def get_services(root: str | Path) -> dict[str, Any]:
 93        if not has_pyproject_toml(root):
 94            return {}
 95
 96        with open(Path(root, "pyproject.toml"), "rb") as f:
 97            pyproject = tomllib.load(f)
 98
 99        return (
100            pyproject.get("tool", {})
101            .get("plain", {})
102            .get("dev", {})
103            .get("services", {})
104        )
105
106    def run(self) -> None:
107        if not self.acquire():
108            click.secho(self.already_running_message(self.read_pidfile()), fg="yellow")
109            return
110
111        self.prepare_log()
112        self.init_poncho(print)
113
114        assert self.poncho is not None, "poncho should be initialized"
115
116        try:
117            services = self.get_services(APP_PATH.parent)
118            for name, data in services.items():
119                env = {
120                    **os.environ,
121                    "PYTHONUNBUFFERED": "true",
122                    "FORCE_COLOR": "1",
123                    **data.get("env", {}),
124                }
125                self.poncho.add_process(name, data["cmd"], env=env)
126
127            self.poncho.loop()
128        finally:
129            self.release()
130            self.close()