furtka/furtka/install_runner.py

"""Background job for app installs — progress-visible via state file.

The slow part of installing an app is `docker compose pull` on a large
image (Jellyfin ~500 MB); without progress feedback, the UI modal sits
dead on "Installing…" for 30+ seconds and the user wonders if it hung.

This module mirrors the exact same shape as ``furtka.catalog`` and
``furtka.updater`` so the UI can poll an install just like it polls a
catalog sync or a self-update. The split is:

- ``furtka.api._do_install`` runs synchronously: resolve source(s), copy
  the app folder(s), write .env. Those are fast, and their failures
  deserve an immediate 4xx so the install modal can surface them in-line.
- After that the API writes an initial state file (stage
  "pulling_image") and dispatches ``systemd-run --unit=furtka-install-
  <name>`` to run ``furtka app install-bg <name>`` in the background.
  That CLI subcommand is what calls ``run_install()`` here — it does the
  docker-facing phases and writes state transitions as it goes.

If the API also wrote a plan file at ``/var/lib/furtka/install-plan.json``
(because the target had transitive dependencies), the runner iterates
through every app in ``to_install`` — pulling, creating volumes, firing
``on_install`` hooks against already-up providers, then ``compose up`` —
so providers are ready before consumers' hooks try to talk to them. The
state file's ``target`` field carries the original user-chosen app name
so the UI can show "Installing mosquitto (required by zigbee2mqtt)".

State file schema (``/var/lib/furtka/install-state.json``):

    {
      "stage": "pulling_image" | "creating_volumes"
             | "running_hooks" | "starting_container" | "done" | "error",
      "updated_at": "2026-04-21T17:30:45+0200",
      "app": "mosquitto",            // app currently being processed
      "target": "zigbee2mqtt",       // original target (== app for single-app installs)
      "version": "1.0.0",            // added at "done"
      "error": "details..."          // added at "error"
    }

Lock: ``/run/furtka/install.lock`` (tmpfs, reboot-safe). Global, not
per-app — two parallel installs are not a v1 use-case and the lock
keeps the state-file representation simple (one in-flight install at
a time).
"""

from __future__ import annotations

import fcntl
import json
import os
import re
import time
from pathlib import Path

from furtka import deps, dockerops, installer
from furtka.manifest import SETTING_NAME_RE, Manifest, load_manifest
from furtka.paths import apps_dir

_INSTALL_STATE = Path(os.environ.get("FURTKA_INSTALL_STATE", "/var/lib/furtka/install-state.json"))
_INSTALL_PLAN = Path(os.environ.get("FURTKA_INSTALL_PLAN", "/var/lib/furtka/install-plan.json"))
_LOCK_PATH = Path(os.environ.get("FURTKA_INSTALL_LOCK", "/run/furtka/install.lock"))

_ON_INSTALL_TIMEOUT_SECONDS = 60.0
_FURTKA_JSON_RE = re.compile(r"^FURTKA_JSON:\s*(.*)$")


class InstallRunnerError(RuntimeError):
    """Any failure in the background install flow that should surface to the caller."""


def state_path() -> Path:
    return _INSTALL_STATE


def plan_path() -> Path:
    return _INSTALL_PLAN


def lock_path() -> Path:
    return _LOCK_PATH


def write_state(stage: str, **extra) -> None:
    """Atomic JSON state write — same shape as catalog/update-state."""
    state_path().parent.mkdir(parents=True, exist_ok=True)
    tmp = state_path().with_suffix(".tmp")
    payload = {"stage": stage, "updated_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"), **extra}
    tmp.write_text(json.dumps(payload, indent=2))
    tmp.replace(state_path())


def read_state() -> dict:
    try:
        data = json.loads(state_path().read_text())
    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


TERMINAL_STAGES = frozenset({"done", "error"})

# How old an unfinished state must be before we believe the job is gone.
# Covers the gap between the API writing the first stage and the detached
# runner taking the lock.
_STALE_AFTER_SECONDS = 60


def clear_stale_state(now: float | None = None) -> bool:
    """Mark an unfinished install as failed if no runner is alive any more.

    install-state.json lives in /var/lib and survives reboots, OOM kills and
    a runner that never started — each of which used to block every later
    install with "another install is in progress" until someone fixed it
    over SSH. The runner holds the flock for its whole run, so a free lock
    plus an old timestamp means nobody is coming back to finish. Returns
    True when it reset the state.
    """
    state = read_state()
    stage = state.get("stage", "")
    if not stage or stage in TERMINAL_STAGES:
        return False
    try:
        updated = time.mktime(time.strptime(state.get("updated_at", ""), "%Y-%m-%dT%H:%M:%S%z"))
    except (TypeError, ValueError):
        updated = 0.0
    if (now if now is not None else time.time()) - updated < _STALE_AFTER_SECONDS:
        return False
    try:
        fh = acquire_lock()
    except (InstallRunnerError, OSError):
        return False  # a runner is still working
    try:
        write_state(
            "error",
            app=state.get("app", ""),
            error=f"the install was interrupted at '{stage}' (restart or crash) — try again",
        )
    finally:
        fh.close()
    return True


def _read_plan(target: str) -> dict:
    """Load the install plan if the API wrote one; otherwise the single-app fallback.

    The plan file is consumed once — removed after read so a stale plan from
    a previous install can't accidentally steer this run. If the file is
    missing/unparseable we synthesize a one-element plan from the target arg
    so the old single-app behaviour still works (CLI invocations, smoke tests).
    """
    try:
        raw = plan_path().read_text()
    except (FileNotFoundError, OSError):
        return {"target": target, "to_install": [target]}
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return {"target": target, "to_install": [target]}
    finally:
        try:
            plan_path().unlink()
        except OSError:
            pass
    if not isinstance(data, dict):
        return {"target": target, "to_install": [target]}
    return {
        "target": data.get("target", target),
        "to_install": data.get("to_install") or [target],
    }


def acquire_lock():
    path = lock_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    fh = path.open("w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError as e:
        fh.close()
        raise InstallRunnerError("another install is already in progress") from e
    return fh


def _parse_hook_output(text: str) -> dict[str, str]:
    """Extract KEY=VALUE pairs from hook stdout plus any FURTKA_JSON: {...} line.

    KEY=VALUE keys must match the manifest's SETTING_NAME regex (UPPER_SNAKE_CASE)
    so a misbehaving hook can't inject e.g. `PATH=` and clobber the container's
    runtime environment.

    The FURTKA_JSON sentinel is opt-in for hooks that need to return structured
    data later (e.g. a list of generated certificates). Only string values are
    accepted; non-string values raise so a hook can't smuggle non-env content
    into the .env file. JSON values overlay KEY=VALUE values.
    """
    out: dict[str, str] = {}

    # First pass: skip FURTKA_JSON lines for KEY=VALUE extraction.
    kv_lines = [line for line in text.splitlines() if not _FURTKA_JSON_RE.match(line.strip())]
    kv = installer.parse_env_text("\n".join(kv_lines))
    for key, value in kv.items():
        if not SETTING_NAME_RE.match(key):
            raise InstallRunnerError(
                f"hook returned invalid env-var name {key!r} "
                "(must be UPPER_SNAKE_CASE, e.g. MQTT_USER)"
            )
        out[key] = value

    # Second pass: pick up FURTKA_JSON sentinels.
    for raw in text.splitlines():
        m = _FURTKA_JSON_RE.match(raw.strip())
        if not m:
            continue
        try:
            payload = json.loads(m.group(1))
        except json.JSONDecodeError as e:
            raise InstallRunnerError(f"hook returned invalid FURTKA_JSON payload: {e}") from e
        if not isinstance(payload, dict):
            raise InstallRunnerError(
                "hook FURTKA_JSON payload must be an object of KEY=VALUE strings"
            )
        for key, value in payload.items():
            if not isinstance(key, str) or not SETTING_NAME_RE.match(key):
                raise InstallRunnerError(f"hook FURTKA_JSON key {key!r} must be UPPER_SNAKE_CASE")
            if not isinstance(value, str):
                raise InstallRunnerError(f"hook FURTKA_JSON value for {key!r} must be a string")
            out[key] = value
    return out


def _merge_hook_output_into_env(env_path: Path, hook_stdout: str) -> None:
    """Overlay hook-returned keys onto an app's `.env`. Hook wins on conflict.

    Re-runs the placeholder-secret check so a hook returning literal "changeme"
    is refused the same way an unedited .env.example is. Re-chmods to 0600 so
    even an interrupted run leaves the file root-only.
    """
    overlay = _parse_hook_output(hook_stdout)
    if not overlay:
        return
    existing = installer.read_env_values(env_path)
    merged: dict[str, str] = {}
    merged.update(existing)
    merged.update(overlay)  # hook wins
    installer.write_env(env_path, merged)
    env_path.chmod(0o600)
    bad = installer._placeholder_keys(env_path)
    if bad:
        raise InstallRunnerError(
            f"{env_path}: hook returned placeholder values for {', '.join(bad)}"
        )


def _fire_install_hooks(consumer: Manifest, consumer_dir: Path) -> None:
    """Run each `on_install` hook against the corresponding provider's container.

    The provider must already be running (its `compose up` ran earlier in the
    same plan). Hook stdout is parsed via `_parse_hook_output` and merged into
    the consumer's `.env` before its own `compose up` fires.
    """
    for req in consumer.requires:
        if not req.on_install:
            continue
        provider_dir = apps_dir() / req.app
        provider_manifest_path = provider_dir / "manifest.json"
        if not provider_manifest_path.is_file():
            raise InstallRunnerError(f"{consumer.name}: required app {req.app!r} is not installed")
        # Validate provider manifest loads (matches the contract the rest of
        # the system relies on — never trust a provider folder with a busted
        # manifest).
        load_manifest(provider_manifest_path, expected_name=req.app)
        hook_abs = provider_dir / req.on_install
        if not hook_abs.is_file():
            raise InstallRunnerError(
                f"{consumer.name}: on_install hook {req.on_install!r} missing in provider {req.app}"
            )
        service = deps.provider_exec_service(provider_dir, req.app)
        stdout = dockerops.compose_exec_script(
            provider_dir,
            req.app,
            service,
            hook_abs,
            env={
                "FURTKA_CONSUMER_APP": consumer.name,
                "FURTKA_CONSUMER_VERSION": consumer.version,
            },
            timeout=_ON_INSTALL_TIMEOUT_SECONDS,
        )
        _merge_hook_output_into_env(consumer_dir / ".env", stdout)


def run_install(name: str) -> None:
    """Docker-facing phases of the install: pull → volumes → hooks → compose up.

    Called by the ``furtka app install-bg <name>`` CLI subcommand from the
    systemd-run spawned by the API. Assumes the API has already run
    ``installer.install_from()`` for every app in the plan, so each app folder,
    `.env`, and manifest are on disk under ``apps_dir() / <name>``.

    If ``/var/lib/furtka/install-plan.json`` exists, every app in its
    ``to_install`` is processed in order (providers before consumers). Each
    provider is fully up before the consumer's ``on_install`` hooks fire,
    so a hook can ``mosquitto_passwd``/`createuser` against a live broker/DB.

    Every phase transition is written to the state file for the UI to poll.
    On exception the state flips to ``"error"`` with the message, then the
    exception is re-raised so the CLI exits non-zero and journald gets a
    traceback. Per-app failure aborts the rest of the plan: a half-installed
    consumer whose provider is fine is recoverable by retrying.
    """
    with acquire_lock():
        plan = _read_plan(name)
        target = plan["target"]
        to_install = list(plan["to_install"])
        try:
            last_manifest = None
            for app_name in to_install:
                target_dir = apps_dir() / app_name
                m = load_manifest(target_dir / "manifest.json", expected_name=app_name)
                last_manifest = m
                write_state("pulling_image", app=app_name, target=target)
                dockerops.compose_pull(target_dir, app_name)
                write_state("creating_volumes", app=app_name, target=target)
                for short in m.volumes:
                    dockerops.ensure_volume(m.volume_name(short))
                if m.requires:
                    write_state("running_hooks", app=app_name, target=target)
                    _fire_install_hooks(m, target_dir)
                write_state("starting_container", app=app_name, target=target)
                dockerops.compose_up(target_dir, app_name)
            # Terminal state carries the original target's name + version so
            # the UI's poll loop ("is install of <target> done yet?") still
            # works unchanged.
            if last_manifest is not None and last_manifest.name == target:
                write_state("done", app=target, target=target, version=last_manifest.version)
            else:
                # Fallback: target wasn't last in the plan (shouldn't happen for
                # a well-formed plan, but don't crash on the terminal write).
                write_state("done", app=target, target=target)
        except Exception as e:
            current = read_state().get("app", target)
            write_state("error", app=current, target=target, error=str(e))
            raise