#!/usr/bin/env python3

"""Read and atomically update Panama's private Home Assistant settings.

Secret values are accepted only as a single JSON object on stdin and are never
returned. The command line therefore remains safe to inspect with ps(1).
"""

from __future__ import annotations

import fcntl
import json
import os
import pathlib
import re
import shlex
import sys
import tempfile
import urllib.parse
from collections.abc import Mapping, Sequence
from typing import Any


KEY_URL = "PANAMA_HOME_ASSISTANT_URL"
KEY_TOKEN = "PANAMA_HOME_ASSISTANT_TOKEN"
KEY_ENTITIES = "PANAMA_HOME_ASSISTANT_ENTITIES"
TARGET_KEYS = (KEY_URL, KEY_TOKEN, KEY_ENTITIES)
ASSIGNMENT = re.compile(
    r"^(?P<prefix>\s*(?:export\s+)?)(?P<key>[A-Za-z_][A-Za-z0-9_]*)\s*=(?P<value>.*)$"
)
ENTITY_ID = re.compile(r"^[a-z_]+\.[a-z0-9_]+$")
# Personal environment lives OUTSIDE the checkout (~/.config/panama/env),
# where agents, backup tools and `panama update` never walk. The in-repo
# config/bash/env is only the legacy location for unmigrated machines — and
# writing there once rebuilt a file the owner thought was hand-maintained.
LEGACY_ENV = pathlib.Path(__file__).resolve().parents[3] / "bash/env"


def default_env() -> pathlib.Path:
    config_home = pathlib.Path(
        os.environ.get("XDG_CONFIG_HOME", str(pathlib.Path.home() / ".config")))
    migrated = config_home / "panama" / "env"
    if migrated.exists() or not LEGACY_ENV.exists():
        return migrated
    return LEGACY_ENV


class ConfigError(RuntimeError):
    """An error code safe to display without including submitted values."""


def env_path() -> pathlib.Path:
    override = os.environ.get("PANAMA_HOME_ASSISTANT_ENV_FILE", "")
    return pathlib.Path(override) if override else default_env()


def parse_assignment(raw: str) -> str | None:
    try:
        parsed = shlex.split(raw, comments=True, posix=True)
    except ValueError:
        return None
    return parsed[0] if len(parsed) == 1 else None


def read_values(path: pathlib.Path) -> dict[str, str]:
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except FileNotFoundError:
        return {}
    except OSError as error:
        raise ConfigError("read-failed") from error

    values: dict[str, str] = {}
    for line in lines:
        match = ASSIGNMENT.match(line)
        if not match or match.group("key") not in TARGET_KEYS:
            continue
        value = parse_assignment(match.group("value"))
        if value is not None:
            values[match.group("key")] = value
    return values


def normalize_url(value: Any) -> str:
    if not isinstance(value, str):
        raise ConfigError("invalid-url")
    normalized = value.strip().rstrip("/")
    if not normalized:
        return ""
    parsed = urllib.parse.urlsplit(normalized)
    if (
        parsed.scheme not in {"http", "https"}
        or not parsed.hostname
        or parsed.username
        or parsed.password
    ):
        raise ConfigError("invalid-url")
    return normalized


def normalize_token(value: Any) -> str:
    if not isinstance(value, str) or "\x00" in value or "\n" in value or "\r" in value:
        raise ConfigError("invalid-token")
    return value.strip()


def normalize_entities(value: Any) -> tuple[str, ...]:
    if isinstance(value, str):
        candidates: Sequence[Any] = re.split(r"[,\n]", value)
    elif isinstance(value, list):
        candidates = value
    else:
        raise ConfigError("invalid-entities")

    entities: list[str] = []
    for candidate in candidates:
        if not isinstance(candidate, str):
            raise ConfigError("invalid-entities")
        entity_id = candidate.strip()
        if not entity_id:
            continue
        if not ENTITY_ID.fullmatch(entity_id):
            raise ConfigError("invalid-entities")
        if entity_id not in entities:
            entities.append(entity_id)
    return tuple(entities)


def public_state(values: Mapping[str, str]) -> dict[str, object]:
    try:
        url = normalize_url(values.get(KEY_URL, ""))
        entities = list(normalize_entities(values.get(KEY_ENTITIES, "")))
        error = ""
    except ConfigError as config_error:
        url = ""
        entities = []
        error = str(config_error)
    token_configured = bool(values.get(KEY_TOKEN, "").strip())
    return {
        "ok": error == "",
        "configured": bool(url and token_configured),
        "tokenConfigured": token_configured,
        "url": url,
        "entities": entities,
        "error": error,
    }


def render_updated(original: str, updates: Mapping[str, str]) -> str:
    lines = original.splitlines(keepends=True)
    rendered: list[str] = []
    replaced: set[str] = set()

    for line in lines:
        content = line.rstrip("\r\n")
        ending = line[len(content) :]
        match = ASSIGNMENT.match(content)
        key = match.group("key") if match else ""
        if key not in updates:
            rendered.append(line)
            continue
        if key in replaced:
            continue
        rendered.append(f"export {key}={shlex.quote(updates[key])}{ending or os.linesep}")
        replaced.add(key)

    if rendered and not rendered[-1].endswith(("\n", "\r")):
        rendered[-1] += os.linesep
    for key in TARGET_KEYS:
        if key in updates and key not in replaced:
            rendered.append(f"export {key}={shlex.quote(updates[key])}{os.linesep}")
    return "".join(rendered)


def atomic_write(path: pathlib.Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary_name = tempfile.mkstemp(prefix=".env.", dir=path.parent)
    temporary = pathlib.Path(temporary_name)
    try:
        os.fchmod(descriptor, 0o600)
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        os.chmod(path, 0o600)
        directory_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(directory_fd)
        finally:
            os.close(directory_fd)
    finally:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass


def read_payload() -> dict[str, Any]:
    line = sys.stdin.buffer.readline(1_048_577)
    if not line or len(line) > 1_048_576:
        raise ConfigError("invalid-payload")
    try:
        payload = json.loads(line)
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ConfigError("invalid-payload") from error
    if not isinstance(payload, dict):
        raise ConfigError("invalid-payload")
    return payload


def write_payload(path: pathlib.Path, payload: Mapping[str, Any]) -> dict[str, object]:
    allowed = {"url", "token", "entities"}
    if not set(payload).issubset(allowed) or not payload:
        raise ConfigError("invalid-payload")

    updates: dict[str, str] = {}
    if "url" in payload:
        updates[KEY_URL] = normalize_url(payload["url"])
    if "token" in payload:
        updates[KEY_TOKEN] = normalize_token(payload["token"])
    if "entities" in payload:
        updates[KEY_ENTITIES] = ",".join(normalize_entities(payload["entities"]))

    lock_path = path.with_name("." + path.name + ".lock")
    path.parent.mkdir(parents=True, exist_ok=True)
    lock_fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        os.fchmod(lock_fd, 0o600)
        fcntl.flock(lock_fd, fcntl.LOCK_EX)
        try:
            original = path.read_text(encoding="utf-8")
        except FileNotFoundError:
            original = ""
        except OSError as error:
            raise ConfigError("read-failed") from error
        atomic_write(path, render_updated(original, updates))
    finally:
        os.close(lock_fd)

    result = public_state(read_values(path))
    result["ok"] = True
    result["error"] = ""
    return result


def compact_json(value: Mapping[str, object]) -> str:
    return json.dumps(value, separators=(",", ":"))


def main() -> None:
    command = sys.argv[1] if len(sys.argv) > 1 else "status"
    if len(sys.argv) > 2 or command not in {"status", "write"}:
        raise ConfigError("usage")
    path = env_path()
    result = public_state(read_values(path)) if command == "status" else write_payload(path, read_payload())
    print(compact_json(result))


if __name__ == "__main__":
    try:
        main()
    except ConfigError as error:
        print(compact_json({"ok": False, "error": str(error)}))
        raise SystemExit(1) from error
    except OSError:
        print(compact_json({"ok": False, "error": "write-failed"}))
        raise SystemExit(1)