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