"""Generic user-decision IPC for Swift app confirmation flows."""
from __future__ import annotations

import json
import os
import threading
import time
from typing import Any

from persistence import generate_secure_token, secure_atomic_write
from tools._logging import should_stop_tool

_MUGI_DIR_IPC = os.path.abspath(os.path.expanduser(os.environ.get("MUGI_DATA_DIR", "~/.mugi")))
if os.path.isdir(_MUGI_DIR_IPC):
    try:
        _ipc_stat = os.stat(_MUGI_DIR_IPC)
        if _ipc_stat.st_uid != os.getuid():
            _MUGI_DIR_IPC = os.path.expanduser("~/.mugi")
    except OSError:
        _MUGI_DIR_IPC = os.path.expanduser("~/.mugi")

_USER_DECISION_REQUEST_PATH = os.path.join(_MUGI_DIR_IPC, ".user_decision_request.json")
_USER_DECISION_RESPONSE_PATH = os.path.join(_MUGI_DIR_IPC, ".user_decision_response.json")
_LEGACY_YTDL_REQUEST_PATH = os.path.join(_MUGI_DIR_IPC, ".ytdl_cookies_request.json")
_LEGACY_YTDL_RESPONSE_PATH = os.path.join(_MUGI_DIR_IPC, ".ytdl_cookies_response.json")

_pending_lock = threading.Lock()
_pending_by_id: dict[str, dict[str, Any]] = {}

_KIND_SPECS: dict[str, dict[str, Any]] = {
    "browser_cookie_source": {
        "choices": ("cancel", "safari", "chrome", "firefox", "brave"),
        "max_context_bytes": 16 * 1024,
        "max_title_chars": 120,
        "max_message_chars": 400,
        "max_url_chars": 2000,
    }
}


def allowed_user_decision_kinds() -> set[str]:
    return set(_KIND_SPECS.keys())


def _cleanup_legacy_cookie_files() -> None:
    for path in (_LEGACY_YTDL_REQUEST_PATH, _LEGACY_YTDL_RESPONSE_PATH):
        try:
            if os.path.isfile(path):
                os.unlink(path)
        except OSError:
            pass


def _normalize_source(source: str) -> str:
    src = (source or "").strip().lower()
    if src == "core":
        return "core"
    if src.startswith("plugin:"):
        slug = src.split(":", 1)[1].strip().replace(" ", "_")
        if slug:
            return f"plugin:{slug}"
    raise ValueError("Invalid user-decision source.")


def _sanitize_text(value: Any, max_len: int) -> str:
    text = str(value or "").strip()
    text = text.replace("\r", " ").replace("\n", " ")
    text = text.replace("<", "").replace(">", "")
    return text[:max_len]


def _sanitize_context(kind: str, context: dict | None) -> dict[str, Any]:
    spec = _KIND_SPECS.get(kind)
    if not spec:
        raise ValueError(f"Unknown user-decision kind: {kind}")
    raw = context if isinstance(context, dict) else {}
    clean: dict[str, Any] = {}
    if "title" in raw:
        clean["title"] = _sanitize_text(raw.get("title"), int(spec.get("max_title_chars", 120)))
    if "message" in raw:
        clean["message"] = _sanitize_text(raw.get("message"), int(spec.get("max_message_chars", 400)))
    if "url" in raw:
        clean["url"] = _sanitize_text(raw.get("url"), int(spec.get("max_url_chars", 2000)))
    if "out_template" in raw:
        clean["out_template"] = _sanitize_text(raw.get("out_template"), 512)
    if "format_preference" in raw:
        clean["format_preference"] = _sanitize_text(raw.get("format_preference"), 120)
    if "audio_only" in raw:
        clean["audio_only"] = bool(raw.get("audio_only"))
    encoded = json.dumps(clean, ensure_ascii=False)
    if len(encoded.encode("utf-8")) > int(spec.get("max_context_bytes", 16 * 1024)):
        raise ValueError("User-decision context is too large.")
    return clean


def _validate_envelope(data: Any) -> dict[str, Any] | None:
    if not isinstance(data, dict):
        return None
    request_id = str(data.get("request_id") or "").strip()
    kind = str(data.get("kind") or "").strip()
    source = str(data.get("source") or "").strip()
    if not request_id or kind not in _KIND_SPECS:
        return None
    try:
        source = _normalize_source(source)
    except ValueError:
        return None
    context = _sanitize_context(kind, data.get("context") if isinstance(data.get("context"), dict) else {})
    created_ms_raw = data.get("created_ms")
    try:
        created_ms = int(created_ms_raw)
    except Exception:
        return None
    thread_id = str(data.get("thread_id") or "").strip() or None
    return {
        "request_id": request_id,
        "kind": kind,
        "source": source,
        "context": context,
        "created_ms": created_ms,
        "thread_id": thread_id,
    }


def submit_user_decision_request(
    *,
    kind: str,
    context: dict | None = None,
    source: str = "core",
    thread_id: str | None = None,
) -> str:
    """Create a single pending user-decision request and return request_id."""
    _cleanup_legacy_cookie_files()
    source_norm = _normalize_source(source)
    kind_norm = (kind or "").strip()
    if kind_norm not in _KIND_SPECS:
        raise ValueError(f"Unsupported user-decision kind: {kind_norm}")
    ctx = _sanitize_context(kind_norm, context)
    with _pending_lock:
        if os.path.isfile(_USER_DECISION_REQUEST_PATH):
            raise RuntimeError("Another user decision request is already pending.")
        request_id = generate_secure_token(24)
        payload = {
            "request_id": request_id,
            "kind": kind_norm,
            "context": ctx,
            "source": source_norm,
            "created_ms": int(time.time() * 1000),
            "thread_id": (thread_id or "").strip() or None,
        }
        secure_atomic_write(_USER_DECISION_REQUEST_PATH, json.dumps(payload, ensure_ascii=False))
        _pending_by_id[request_id] = payload
    return request_id


def get_pending_user_decision_from_file() -> dict[str, Any] | None:
    """Read and consume pending request for UI polling."""
    _cleanup_legacy_cookie_files()
    if not os.path.isfile(_USER_DECISION_REQUEST_PATH):
        return None
    try:
        with open(_USER_DECISION_REQUEST_PATH, "r", encoding="utf-8") as f:
            raw = json.load(f)
    except Exception:
        raw = None
    try:
        os.unlink(_USER_DECISION_REQUEST_PATH)
    except OSError:
        pass
    req = _validate_envelope(raw)
    if req is None:
        return None
    with _pending_lock:
        _pending_by_id[req["request_id"]] = req
    return req


def _valid_choice(kind: str, choice: str) -> bool:
    spec = _KIND_SPECS.get(kind) or {}
    allowed = tuple(spec.get("choices") or ())
    return (choice or "").strip().lower() in allowed


def complete_user_decision_file(request_id: str, choice: str, *, kind: str | None = None) -> bool:
    """Persist user response for a known request_id."""
    rid = (request_id or "").strip()
    sel = (choice or "").strip().lower()
    if not rid:
        return False
    with _pending_lock:
        req = _pending_by_id.get(rid)
    kind_from_state = str((req or {}).get("kind") or "").strip()
    if kind is not None and kind.strip():
        posted_kind = kind.strip()
        if posted_kind not in _KIND_SPECS:
            return False
        if kind_from_state and posted_kind != kind_from_state:
            return False
        kind_from_state = posted_kind
    if not kind_from_state or not _valid_choice(kind_from_state, sel):
        return False
    try:
        secure_atomic_write(
            _USER_DECISION_RESPONSE_PATH,
            json.dumps({"request_id": rid, "choice": sel}, ensure_ascii=False),
        )
    except Exception:
        return False
    return True


def wait_user_decision_response(request_id: str, timeout_s: float = 300.0) -> str | None:
    """Wait for response to request_id until timeout/stop."""
    rid = (request_id or "").strip()
    if not rid:
        return None
    deadline = time.time() + max(0.5, float(timeout_s))
    while time.time() < deadline:
        if should_stop_tool():
            return None
        if os.path.isfile(_USER_DECISION_RESPONSE_PATH):
            try:
                with open(_USER_DECISION_RESPONSE_PATH, "r", encoding="utf-8") as f:
                    resp = json.load(f)
            except Exception:
                resp = None
            if isinstance(resp, dict) and (resp.get("request_id") or "").strip() == rid:
                choice = (resp.get("choice") or "").strip().lower()
                with _pending_lock:
                    req = _pending_by_id.get(rid, {})
                kind = str(req.get("kind") or "").strip()
                if kind and _valid_choice(kind, choice):
                    try:
                        os.unlink(_USER_DECISION_RESPONSE_PATH)
                    except OSError:
                        pass
                    with _pending_lock:
                        _pending_by_id.pop(rid, None)
                    return choice
        time.sleep(0.2)
    with _pending_lock:
        req = _pending_by_id.pop(rid, None)
    try:
        if req and os.path.isfile(_USER_DECISION_REQUEST_PATH):
            with open(_USER_DECISION_REQUEST_PATH, "r", encoding="utf-8") as f:
                still = json.load(f)
            if (still.get("request_id") or "").strip() == rid:
                os.unlink(_USER_DECISION_REQUEST_PATH)
    except Exception:
        pass
    return None


def request_user_decision(
    *,
    kind: str,
    context: dict | None = None,
    source: str = "core",
    thread_id: str | None = None,
    timeout_s: float = 300.0,
) -> str | None:
    """Submit and wait for a user-decision request."""
    try:
        req_id = submit_user_decision_request(
            kind=kind,
            context=context,
            source=source,
            thread_id=thread_id,
        )
    except Exception:
        return None
    return wait_user_decision_response(req_id, timeout_s=timeout_s)
