dev.py

#!/usr/bin/env python3
"""SoulSync development launcher.

Starts the backend and Vite dev server together, restarts the backend when
backend source files change, and handles shutdown cleanly across platforms.
"""

from __future__ import annotations

import atexit
import os
import shutil
import signal
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT_DIR = Path(__file__).resolve().parent
LOG_DIR = ROOT_DIR / 'logs'
GUNICORN_CONFIG = ROOT_DIR / 'gunicorn.dev.conf.py'
VITE_URL = os.environ.get('SOULSYNC_WEBUI_VITE_URL', 'http://127.0.0.1:5173').rstrip('/')
# Which interface Vite binds. Loopback by default; `--lan` widens it (see main()).
VITE_BIND_HOST = '127.0.0.1'
VITE_LOG_FILE = Path(os.environ.get('SOULSYNC_WEBUI_VITE_LOG', str(LOG_DIR / 'webui-vite.log')))

INCLUDED_SUFFIXES = {'.py', '.html', '.jinja', '.jinja2'}
SHUTDOWN_GRACE_SECONDS = int(os.environ.get('SOULSYNC_SHUTDOWN_GRACE_SECONDS', '10'))
FORCE_KILL_ON_SHUTDOWN = os.environ.get('SOULSYNC_FORCE_KILL_ON_SHUTDOWN', '1').lower() in {
    '1',
    'true',
    'yes',
    'on',
}

shutdown_requested = False
managed_processes: list[tuple[str, subprocess.Popen, object | None]] = []


def detect_lan_ip() -> str | None:
    """Best-effort address other devices on the network can reach us at.

    Opens a UDP socket toward a public address and reads back the local end the
    OS picked. No packets are actually sent, and nothing needs to be reachable.
    Under WSL this returns the WSL adapter's address, which a phone cannot reach
    — that is what `--lan=<ip>` is for.
    """
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
            probe.connect(('8.8.8.8', 80))
            return probe.getsockname()[0]
    except OSError:
        return None


def resolve_command(*candidates: str) -> str | None:
    for candidate in candidates:
        resolved = shutil.which(candidate)
        if resolved:
            return resolved
    return None


def is_excluded(path: Path) -> bool:
    try:
        relative = path.relative_to(ROOT_DIR)
    except ValueError:
        return False

    parts = relative.parts
    if not parts:
        return False

    if any(part == '__pycache__' for part in parts):
        return True
    if parts[0] in {'.git', 'logs'}:
        return True
    if len(parts) >= 2 and parts[0] == 'webui' and parts[1] == 'node_modules':
        return True
    if len(parts) >= 3 and parts[0] == 'webui' and parts[1] == 'static' and parts[2] == 'dist':
        return True
    return False


def build_backend_env(direct_mode: bool) -> dict[str, str]:
    env = os.environ.copy()
    env.setdefault('SOULSYNC_WEB_DEV_NO_CACHE', '1')
    env.setdefault('SOULSYNC_WEBUI_VITE_DEV', '1')
    env.setdefault('SOULSYNC_WEBUI_VITE_URL', VITE_URL)
    env.setdefault('SOULSYNC_WEBUI_VITE_LOG', str(VITE_LOG_FILE))
    env.setdefault('SOULSYNC_CONFIG_PATH', str(ROOT_DIR / 'config' / 'config.json'))

    if direct_mode:
        env.setdefault('SOULSYNC_WEB_BIND_HOST', '127.0.0.1')
        env.setdefault('SOULSYNC_WEB_BIND_PORT', '8008')

    return env


def start_process(label: str, cmd: list[str], *, log_file: Path | None = None, env: dict[str, str] | None = None) -> tuple[subprocess.Popen, object | None]:
    log_handle = None
    stdout = None
    stderr = None
    if log_file is not None:
        log_file.parent.mkdir(parents=True, exist_ok=True)
        log_handle = log_file.open('ab')
        stdout = log_handle
        stderr = log_handle

    creationflags = 0
    start_new_session = False
    if os.name == 'nt':
        creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
    else:
        start_new_session = True

    try:
        proc = subprocess.Popen(
            cmd,
            cwd=str(ROOT_DIR),
            env=env,
            stdin=subprocess.DEVNULL,
            stdout=stdout,
            stderr=stderr,
            creationflags=creationflags,
            start_new_session=start_new_session,
        )
    except Exception:
        if log_handle is not None:
            log_handle.close()
        raise

    managed_processes.append((label, proc, log_handle))
    return proc, log_handle


def wait_for_exit(proc: subprocess.Popen, seconds: int) -> bool:
    checks = max(1, int(seconds * 10))
    for _ in range(checks):
        if proc.poll() is not None:
            return True
        time.sleep(0.1)
    return proc.poll() is not None


def stop_process(label: str, proc: subprocess.Popen, log_handle: object | None) -> None:
    if proc.poll() is not None:
        if log_handle is not None:
            log_handle.close()
        return

    print(f'Stopping {label}...')

    try:
        if os.name == 'nt':
            proc.terminate()
        else:
            os.killpg(proc.pid, signal.SIGTERM)
    except ProcessLookupError:
        pass

    if not wait_for_exit(proc, SHUTDOWN_GRACE_SECONDS):
        if not FORCE_KILL_ON_SHUTDOWN:
            print(f'{label} did not exit in time; skipping forced kill for this test run.')
        else:
            print(f'{label} did not exit in time; forcing shutdown...')
            if os.name == 'nt':
                subprocess.run(
                    ['taskkill', '/T', '/F', '/PID', str(proc.pid)],
                    check=False,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                )
            else:
                try:
                    os.killpg(proc.pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
            wait_for_exit(proc, 5)

    if log_handle is not None:
        log_handle.close()


def cleanup() -> None:
    global shutdown_requested

    if shutdown_requested:
        return
    shutdown_requested = True

    for label, proc, log_handle in reversed(managed_processes):
        stop_process(label, proc, log_handle)


def compute_backend_watch_state() -> str:
    rows: list[str] = []

    for dirpath, dirnames, filenames in os.walk(ROOT_DIR):
        current_dir = Path(dirpath)
        if is_excluded(current_dir):
            dirnames[:] = []
            continue

        dirnames[:] = [
            name
            for name in dirnames
            if not is_excluded(current_dir / name) and name != '__pycache__'
        ]

        for filename in filenames:
            path = current_dir / filename
            if path.suffix not in INCLUDED_SUFFIXES:
                continue
            try:
                stat = path.stat()
            except FileNotFoundError:
                continue
            rows.append(f'{stat.st_mtime_ns} {path}')

    return '\n'.join(sorted(rows))


def start_vite() -> subprocess.Popen:
    npm = resolve_command('npm', 'npm.cmd')
    if npm is None:
        raise SystemExit('npm is required to run the Vite dev server.')

    print(f'Starting Vite dev server at {VITE_URL}...')
    vite_cmd = [
        npm,
        '--prefix',
        str(ROOT_DIR / 'webui'),
        'run',
        'dev',
        '--',
        '--host',
        VITE_BIND_HOST,
        '--port',
        '5173',
    ]
    proc, _ = start_process('Vite dev server', vite_cmd, log_file=VITE_LOG_FILE, env=os.environ.copy())
    return proc


def wait_for_vite_ready(vite_proc: subprocess.Popen) -> None:
    ready_url = f'{VITE_URL}/static/dist/@vite/client'
    vite_ready = False

    for _ in range(50):
        if vite_proc.poll() is not None:
            print('Warning: Vite dev server exited before it became ready.')
            break
        try:
            with urllib.request.urlopen(ready_url, timeout=1) as response:
                if response.status < 400:
                    vite_ready = True
                    break
        except (urllib.error.URLError, TimeoutError, OSError):
            pass
        time.sleep(0.2)

    if vite_ready:
        print('Vite dev server is ready.')
    else:
        print('Warning: timed out waiting for the Vite dev server.')
        print('The backend will still start, but the frontend may not hot-reload yet.')


def start_backend() -> tuple[subprocess.Popen, object | None]:
    backend_mode = os.environ.get('SOULSYNC_DEV_BACKEND', '').strip().lower()
    direct_mode = backend_mode == 'direct'
    gunicorn_mode = backend_mode == 'gunicorn'

    if not backend_mode:
        if os.name == 'nt':
            direct_mode = True
        elif resolve_command('gunicorn') is None:
            print('gunicorn not found; falling back to direct Python server.')
            direct_mode = True
        else:
            gunicorn_mode = True

    print('Starting SoulSync web server...')

    if gunicorn_mode:
        gunicorn = resolve_command('gunicorn')
        if gunicorn is None:
            raise SystemExit('gunicorn is not available but SOULSYNC_DEV_BACKEND=gunicorn was requested.')
        print(f'Using Gunicorn config: {GUNICORN_CONFIG}')
        cmd = [gunicorn, '-c', str(GUNICORN_CONFIG), 'wsgi:application']
    else:
        print('Using direct Python server for backend.')
        cmd = [sys.executable, str(ROOT_DIR / 'web_server.py')]

    proc, log_handle = start_process(
        'SoulSync web server',
        cmd,
        env=build_backend_env(direct_mode),
    )
    return proc, log_handle


def watch_and_run_backend() -> None:
    last_state = compute_backend_watch_state()
    backend_proc, backend_log = start_backend()

    try:
        while not shutdown_requested:
            time.sleep(1)

            if backend_proc.poll() is not None:
                print('SoulSync web server exited. Restarting...')
                stop_process('SoulSync web server', backend_proc, backend_log)
                managed_processes.pop()
                backend_proc, backend_log = start_backend()
                last_state = compute_backend_watch_state()
                continue

            current_state = compute_backend_watch_state()
            if current_state != last_state:
                print('Detected backend file changes. Restarting SoulSync web server...')
                last_state = current_state
                stop_process('SoulSync web server', backend_proc, backend_log)
                managed_processes.pop()
                backend_proc, backend_log = start_backend()
    finally:
        if backend_proc.poll() is None:
            stop_process('SoulSync web server', backend_proc, backend_log)
            if managed_processes:
                managed_processes.pop()


def main() -> int:
    # `--lan` exposes the dev server on the network (binds 0.0.0.0) so you can
    # reach it from another device at http://<this-pc-ip>:8008. Default stays
    # localhost-only. Set before build_backend_env() so both backend modes see it.
    lan_arg = next((a for a in sys.argv if a == '--lan' or a.startswith('--lan=')), None)
    if lan_arg:
        global VITE_URL, VITE_BIND_HOST
        os.environ['SOULSYNC_WEB_BIND_HOST'] = '0.0.0.0'
        # Exposing Flask is only half of it. In dev the page loads React from the
        # VITE dev server by absolute URL, so Vite has to (a) listen off-loopback
        # and (b) be ADVERTISED at an address the other device can reach. Miss
        # either and index.html hands the phone "http://127.0.0.1:5173/…" — which
        # IS the phone — so the bundle never loads and every React page renders
        # empty while the vanilla pages, served by Flask, look perfectly fine.
        VITE_BIND_HOST = '0.0.0.0'
        host = lan_arg.split('=', 1)[1].strip() if '=' in lan_arg else (detect_lan_ip() or '')
        if host:
            VITE_URL = f'http://{host}:5173'
            os.environ['SOULSYNC_WEBUI_VITE_URL'] = VITE_URL
            print(f'LAN mode ON — open http://{host}:8008 from the other device.')
        else:
            print('LAN mode ON, but this machine\'s LAN address could not be detected.')
            print('Re-run as --lan=<this-pc-ip>, or React pages will be blank on other devices.')
        print('Allow ports 8008 and 5173 through the firewall if needed.')

    if not (ROOT_DIR / 'webui' / 'node_modules').is_dir():
        print('webui/node_modules is missing.')
        print('Run: cd webui && npm ci')
        return 1

    vite_proc = start_vite()
    try:
        wait_for_vite_ready(vite_proc)
        print(f'Vite log: {VITE_LOG_FILE}')
        print('Backend file watching is enabled.')
        watch_and_run_backend()
    finally:
        cleanup()
    return 0


def _handle_signal(signum: int, _frame) -> None:
    raise SystemExit(130 if signum == signal.SIGINT else 143)


signal.signal(signal.SIGINT, _handle_signal)
if hasattr(signal, 'SIGTERM'):
    signal.signal(signal.SIGTERM, _handle_signal)
if hasattr(signal, 'SIGBREAK'):
    signal.signal(signal.SIGBREAK, _handle_signal)

atexit.register(cleanup)

if __name__ == '__main__':
    raise SystemExit(main())