#!/usr/bin/env python3
"""GPU-On-Demand-Manager für houston.

Lauscht auf den öffentlichen Ports der GPU-Dienste und startet den
jeweiligen systemd-Dienst erst bei eingehender Verbindung (TCP-Proxy auf
den internen Port). Nach IDLE_STOP_SECONDS ohne aktive Verbindung wird
der Dienst gestoppt, damit er kein VRAM belegt.

Konfiguration unten in SERVICES; rein stdlib (asyncio).
"""
from __future__ import annotations

import asyncio
import logging
import os
import time

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
LOG = logging.getLogger("gpu-ondemand")

IDLE_STOP_SECONDS = int(os.getenv("GPU_ONDEMAND_IDLE", "900"))
START_TIMEOUT_SECONDS = int(os.getenv("GPU_ONDEMAND_START_TIMEOUT", "600"))
LISTEN_HOST = "127.0.0.1"

SERVICES = {
    8088: {"unit": "voice2image.service", "backend_port": 18088},
    8026: {"unit": "ltx-video.service", "backend_port": 18026},
    8015: {"unit": "ace-step.service", "backend_port": 18015},
    8021: {"unit": "tts-clone.service", "backend_port": 18021},
}


class ServiceState:
    def __init__(self, unit: str, backend_port: int) -> None:
        self.unit = unit
        self.backend_port = backend_port
        self.active_connections = 0
        self.last_activity = 0.0
        self.start_lock = asyncio.Lock()


async def systemctl(*args: str) -> int:
    proc = await asyncio.create_subprocess_exec(
        "systemctl", *args,
        stdout=asyncio.subprocess.DEVNULL,
        stderr=asyncio.subprocess.DEVNULL,
    )
    return await proc.wait()


async def is_active(unit: str) -> bool:
    return await systemctl("is-active", "--quiet", unit) == 0


async def backend_ready(state: ServiceState) -> bool:
    try:
        _, writer = await asyncio.wait_for(
            asyncio.open_connection(LISTEN_HOST, state.backend_port), timeout=2.0)
        writer.close()
        await writer.wait_closed()
        return True
    except Exception:
        return False


async def ensure_backend(state: ServiceState) -> bool:
    """Startet den Dienst falls nötig und wartet, bis der Port erreichbar ist."""
    if await backend_ready(state):
        return True
    async with state.start_lock:
        if await backend_ready(state):
            return True
        LOG.info("Starte %s (on demand)…", state.unit)
        await systemctl("start", state.unit)
        deadline = time.monotonic() + START_TIMEOUT_SECONDS
        while time.monotonic() < deadline:
            if await backend_ready(state):
                LOG.info("%s bereit.", state.unit)
                return True
            await asyncio.sleep(1.0)
        LOG.error("%s wurde nicht rechtzeitig bereit.", state.unit)
        return False


async def pipe(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, state: ServiceState) -> None:
    try:
        while True:
            chunk = await reader.read(65536)
            if not chunk:
                break
            state.last_activity = time.monotonic()
            writer.write(chunk)
            await writer.drain()
    except (ConnectionResetError, BrokenPipeError, asyncio.CancelledError):
        pass
    finally:
        try:
            writer.close()
        except Exception:
            pass


async def handle_client(state: ServiceState, client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter) -> None:
    state.active_connections += 1
    state.last_activity = time.monotonic()
    try:
        if not await ensure_backend(state):
            client_writer.close()
            return
        backend_reader, backend_writer = await asyncio.open_connection(LISTEN_HOST, state.backend_port)
        await asyncio.gather(
            pipe(client_reader, backend_writer, state),
            pipe(backend_reader, client_writer, state),
        )
    except Exception as exc:
        LOG.debug("Verbindungsfehler %s: %s", state.unit, exc)
        try:
            client_writer.close()
        except Exception:
            pass
    finally:
        state.active_connections -= 1
        state.last_activity = time.monotonic()


async def janitor(states: list[ServiceState]) -> None:
    while True:
        await asyncio.sleep(60)
        now = time.monotonic()
        for state in states:
            if state.active_connections > 0:
                continue
            if now - state.last_activity < IDLE_STOP_SECONDS:
                continue
            if await is_active(state.unit):
                LOG.info("Stoppe %s (idle seit %.0fs) — VRAM wird freigegeben.",
                         state.unit, now - state.last_activity)
                await systemctl("stop", state.unit)


async def main() -> None:
    states = []
    for port, cfg in SERVICES.items():
        state = ServiceState(cfg["unit"], cfg["backend_port"])
        state.last_activity = time.monotonic()
        states.append(state)
        server = await asyncio.start_server(
            lambda r, w, s=state: handle_client(s, r, w), LISTEN_HOST, port)
        LOG.info("Lausche auf %s:%d → %s (intern :%d)", LISTEN_HOST, port, cfg["unit"], cfg["backend_port"])

    asyncio.create_task(janitor(states))
    while True:
        await asyncio.sleep(3600)


if __name__ == "__main__":
    asyncio.run(main())