svcforge/services/cli/main.py

"""svcforge — the control plane client.

Talks to the API over HTTP and never touches the database. If the CLI could write to
Postgres, every invariant the API enforces — the state machine, the one-transaction create,
AuthZ in the WHERE clause — would have a back door, and the first 3am incident would go
through it. `ClientSettings` below is what keeps that true in practice.
"""

from __future__ import annotations

import sys
import time
import uuid
from enum import StrEnum
from typing import Annotated, Any

import httpx
import typer
from pydantic_settings import BaseSettings, SettingsConfigDict

from svcforge_core.domain.states import InstanceState

app = typer.Typer(help="svcforge control plane client", no_args_is_help=True)

_TERMINAL = {InstanceState.READY, InstanceState.FAILED, InstanceState.DELETED}


class ServiceType(StrEnum):
    """What the catalog offers. Kept as an enum so typer can complete and validate it."""

    ELASTICSEARCH = "elasticsearch"
    REDIS = "redis"
    POSTGRES = "postgres"
    # Small enough to provision on a cluster with no spare memory; see catalog.yaml.
    PODINFO = "podinfo"
    NGINX = "nginx"


class Size(StrEnum):
    SMALL = "small"
    MEDIUM = "medium"


class ClientSettings(BaseSettings):
    """The two values the CLI needs, and nothing else.

    Its own model rather than `svcforge_core.settings.Settings`, which requires
    `SVCFORGE_PG_DSN`: loading that here would refuse to run the CLI without a database URL
    it then never opens, on a laptop that has no reason to hold one.
    """

    model_config = SettingsConfigDict(
        env_prefix="SVCFORGE_", env_file=".env", env_file_encoding="utf-8", extra="ignore", frozen=True
    )

    api_url: str = "http://localhost:8000"
    api_token: str | None = None


def _client() -> httpx.Client:
    settings = ClientSettings()
    headers = {"authorization": f"Bearer {settings.api_token}"} if settings.api_token else {}
    return httpx.Client(base_url=settings.api_url, headers=headers, timeout=10.0)


def _die(msg: str) -> None:
    typer.secho(msg, fg=typer.colors.RED, err=True)
    raise typer.Exit(code=1)


def _check(resp: httpx.Response) -> Any:  # noqa: ANN401 - a decoded JSON body genuinely is Any
    if resp.status_code == 401:
        _die("401 unauthorized: check SVCFORGE_API_TOKEN")
    if resp.status_code == 404:
        _die("404 not found")
    if resp.status_code == 429:
        _die("429 rate limited: slow down")
    if resp.status_code >= 400:
        _die(f"{resp.status_code}: {resp.text[:400]}")
    return resp.json()


def _parse_ttl(ttl: str | None) -> int | None:
    """'7d' -> 7. Only days, because the API only takes days."""
    if ttl is None:
        return None
    if not ttl.endswith("d") or not ttl[:-1].isdigit():
        _die(f"bad --ttl {ttl!r}: expected something like '7d'")
    return int(ttl[:-1])


def _print_table(rows: list[dict[str, Any]]) -> None:
    if not rows:
        typer.echo("(none)")
        return
    cols = ["id", "service_type", "size", "state", "chart_version", "endpoint"]
    widths = {c: max(len(c), *(len(str(r.get(c) or "-")) for r in rows)) for c in cols}
    typer.echo("  ".join(c.ljust(widths[c]) for c in cols))
    typer.echo("  ".join("-" * widths[c] for c in cols))
    for r in rows:
        typer.echo("  ".join(str(r.get(c) or "-").ljust(widths[c]) for c in cols))


@app.command()
def create(
    service_type: Annotated[ServiceType, typer.Argument(help="what to provision")],
    size: Annotated[Size, typer.Option()] = Size.SMALL,
    ttl: Annotated[str | None, typer.Option(help="e.g. 7d")] = None,
    wait: Annotated[bool, typer.Option(help="poll until ready or failed")] = False,
) -> None:
    """Request an instance. Returns as soon as the API accepts it (202)."""
    body: dict[str, Any] = {"service_type": service_type.value, "size": size.value}
    ttl_days = _parse_ttl(ttl)
    if ttl_days is not None:
        body["ttl_days"] = ttl_days

    with _client() as c:
        data = _check(c.post("/v1/instances", json=body))
        typer.echo(f"{data['id']}  {data['state']}")
        if not wait:
            return

        # 202 means "accepted", not "done". Polling is the client's job precisely because
        # the API refused to block on a helm install that takes four minutes.
        instance_id = data["id"]
        deadline = time.monotonic() + 600
        state = data["state"]
        while time.monotonic() < deadline:
            time.sleep(2)
            data = _check(c.get(f"/v1/instances/{instance_id}"))
            if data["state"] != state:
                state = data["state"]
                typer.echo(f"  -> {state}")
            if state in _TERMINAL:
                break
        else:
            _die("timed out waiting; the task may still be running — check `svcforge status`")

    if state == InstanceState.FAILED:
        _die(f"failed: {data.get('error') or 'no error recorded'}")
    typer.echo(f"endpoint: {data.get('endpoint') or '-'}")


@app.command("list")
def list_instances(
    state: Annotated[InstanceState | None, typer.Option(help="filter by state")] = None,
) -> None:
    """List your team's instances."""
    with _client() as c:
        rows = _check(c.get("/v1/instances"))
    if state is not None:
        rows = [r for r in rows if r["state"] == state.value]
    _print_table(rows)


@app.command()
def status(instance_id: Annotated[uuid.UUID, typer.Argument()]) -> None:
    """Show one instance. Exits non-zero if it is failed, so scripts can branch on it."""
    with _client() as c:
        data = _check(c.get(f"/v1/instances/{instance_id}"))
    _print_table([data])
    if data["state"] == InstanceState.FAILED:
        typer.secho(f"error: {data.get('error')}", fg=typer.colors.RED, err=True)
        raise typer.Exit(code=1)


@app.command()
def delete(
    instance_id: Annotated[uuid.UUID, typer.Argument()],
    yes: Annotated[bool, typer.Option("--yes", "-y", help="skip the confirmation")] = False,
) -> None:
    """Deprovision an instance."""
    if not yes and not typer.confirm(f"delete {instance_id}?"):
        raise typer.Abort
    with _client() as c:
        data = _check(c.delete(f"/v1/instances/{instance_id}"))
    typer.echo(f"{data['id']}  {data['state']}")


def main() -> None:  # pragma: no cover - console-script entrypoint
    try:
        app()
    except httpx.ConnectError:
        typer.secho("cannot reach the API: check SVCFORGE_API_URL", fg=typer.colors.RED, err=True)
        sys.exit(1)


if __name__ == "__main__":  # pragma: no cover
    main()