from __future__ import annotations

import json
import time
from urllib import error, request
from urllib.parse import quote

from .models import TaskState, TaskSummary


class MacServiceError(RuntimeError):
    pass


class MacServiceHTTPError(MacServiceError):
    def __init__(self, status_code: int, detail: str, payload: dict | None = None) -> None:
        super().__init__(f"Mac service HTTP {status_code}: {detail}")
        self.status_code = status_code
        self.detail = detail
        self.payload = payload


class MacControlClient:
    def __init__(self, base_url: str, token: str, timeout_seconds: float = 10.0) -> None:
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.timeout_seconds = timeout_seconds

    def get_state(self) -> TaskState:
        return TaskState.model_validate(self._request_json("GET", "/state")["state"])

    def get_task_state(self, task_ref: str) -> TaskState:
        from urllib.parse import quote

        encoded = quote(task_ref, safe="")
        return TaskState.model_validate(self._request_json("GET", f"/task_state?task_ref={encoded}")["state"])

    def list_task_summaries(self) -> list[TaskSummary]:
        payload = self._request_json("GET", "/task_summaries")
        return [TaskSummary.model_validate(item) for item in payload["tasks"]]

    def list_tasks(self) -> str:
        return str(self._request_json("GET", "/tasks")["text"])

    def create_task(self, title: str) -> TaskState:
        return self._state_post("/task/create", {"title": title})

    def switch_task(self, task_id: str) -> TaskState:
        return self._state_post("/task/switch", {"task_id": task_id})

    def continue_task(self, instruction: str | None = None) -> TaskState:
        payload = {"instruction": instruction} if instruction else {}
        return self._state_post("/task/continue", payload)

    def pause_task(self) -> TaskState:
        return self._state_post("/task/pause", {})

    def add_note(self, text: str, author: str = "telegram") -> TaskState:
        return self._state_post("/task/note", {"text": text, "author": author})

    def append_chat_entry(self, author: str, text: str) -> TaskState:
        return self._state_post("/task/chat_entry", {"author": author, "text": text})

    def run_command(self, command: str) -> TaskState:
        return self._state_post("/task/run", {"command": command})

    def set_workdir(self, workdir: str) -> TaskState:
        return self._state_post("/task/set_workdir", {"workdir": workdir})

    def set_title(self, title: str) -> TaskState:
        return self._state_post("/task/set_title", {"title": title})

    def set_agent_enabled(self, enabled: bool) -> TaskState:
        return self._state_post("/task/set_agent", {"enabled": enabled})

    def reject_pending_command(self) -> TaskState:
        return self._state_post("/task/reject", {})

    def confirm_voice_command(self) -> TaskState:
        return self._state_post("/task/confirm_voice", {})

    def cancel_voice_command(self) -> TaskState:
        return self._state_post("/task/cancel_voice", {})

    def reset_state(self) -> TaskState:
        return self._state_post("/task/reset", {})

    def format_recent_notes(self, limit: int = 5) -> str:
        return str(self._request_json("GET", f"/notes?limit={limit}")["text"])

    def format_queue(self, limit: int = 10) -> str:
        return str(self._request_json("GET", f"/queue?limit={limit}")["text"])

    def progress_summary(self) -> str:
        return str(self._request_json("GET", "/progress")["text"])

    def format_command_history(self, limit: int = 5) -> str:
        return str(self._request_json("GET", f"/history?limit={limit}")["text"])

    def pending_count(self) -> int:
        return int(self._request_json("GET", "/pending_count")["count"])

    def approve(self) -> TaskState:
        return self._state_post("/task/approve", {})

    def get_desktop_bridge_status(self) -> dict:
        return self._request_json("GET", self._cache_busted("/desktop/status"))

    def get_desktop_usage(self) -> dict:
        # Usage changes independently from chat state, so always bypass any
        # intermediate cache between the relay API and the Mac service.
        return self._request_json("GET", self._cache_busted("/desktop/usage"))

    def get_desktop_runtime(self, thread_id: str | None = None) -> dict:
        if thread_id:
            from urllib.parse import quote

            encoded = quote(thread_id, safe="")
            return self._request_json("GET", self._cache_busted(f"/desktop/runtime?thread_id={encoded}"))
        return self._request_json("GET", self._cache_busted("/desktop/runtime"))

    def get_desktop_runtime_ready(self, thread_id: str | None = None) -> dict:
        if thread_id:
            from urllib.parse import quote

            encoded = quote(thread_id, safe="")
            return self._request_json("GET", self._cache_busted(f"/desktop/runtime_ready?thread_id={encoded}"))
        return self._request_json("GET", self._cache_busted("/desktop/runtime_ready"))

    def get_desktop_bridge_snapshot(self, limit: int = 500) -> dict:
        return self._request_json("GET", self._cache_busted(f"/desktop/snapshot?limit={int(limit)}"))

    def get_desktop_thread_snapshot(self, thread_id: str, limit: int = 500) -> dict:
        from urllib.parse import quote

        encoded = quote(thread_id, safe="")
        return self._request_json(
            "GET",
            self._cache_busted(f"/desktop/thread_snapshot?thread_id={encoded}&limit={int(limit)}"),
        )

    def read_desktop_file(self, path: str, encoding: str | None = None) -> dict:
        from urllib.parse import quote

        encoded = quote(path, safe="")
        suffix = f"&encoding={quote(encoding, safe='')}" if encoding else ""
        return self._request_json("GET", f"/desktop/file?path={encoded}{suffix}")

    def get_desktop_bridge_messages(self, limit: int = 500) -> dict:
        return self._request_json("GET", self._cache_busted(f"/desktop/messages?limit={int(limit)}"))

    def list_desktop_threads(self) -> dict:
        return self._request_json("GET", self._cache_busted("/desktop/threads"))

    def list_desktop_projects(self, *, force: bool = False) -> dict:
        path = "/desktop/projects?refresh=1" if force else "/desktop/projects"
        return self._request_json("GET", self._cache_busted(path))

    def select_desktop_thread(self, thread_id: str) -> dict:
        return self._request_json("POST", "/desktop/select_thread", {"thread_id": thread_id})

    def open_desktop_thread(self, thread_id: str) -> dict:
        return self._request_json("POST", "/desktop/open_thread", {"thread_id": thread_id})

    def refresh_desktop_thread(self, thread_id: str | None = None, force: bool = True) -> dict:
        payload = {"force": force}
        if thread_id:
            payload["thread_id"] = thread_id
        return self._request_json("POST", "/desktop/refresh_thread", payload)

    def create_desktop_thread(
        self,
        cwd: str | None = None,
        model: str | None = None,
        effort: str | None = None,
        permission: str | None = None,
    ) -> dict:
        payload = {"cwd": cwd} if cwd else {}
        if model:
            payload["model"] = model
        if effort:
            payload["effort"] = effort
        if permission:
            payload["permission"] = permission
        return self._request_json("POST", "/desktop/new_thread", payload)

    def rename_desktop_thread(self, thread_id: str, title: str) -> dict:
        return self._request_json("POST", "/desktop/rename_thread", {"thread_id": thread_id, "title": title})

    def archive_desktop_thread(self, thread_id: str) -> dict:
        return self._request_json("POST", "/desktop/archive_thread", {"thread_id": thread_id})

    def unarchive_desktop_thread(self, thread_id: str) -> dict:
        return self._request_json("POST", "/desktop/unarchive_thread", {"thread_id": thread_id})

    def pin_desktop_thread(self, thread_id: str) -> dict:
        return self._request_json("POST", "/desktop/pin_thread", {"thread_id": thread_id})

    def unpin_desktop_thread(self, thread_id: str) -> dict:
        return self._request_json("POST", "/desktop/unpin_thread", {"thread_id": thread_id})

    def fork_desktop_thread(self, thread_id: str, cwd: str | None = None) -> dict:
        payload = {"thread_id": thread_id}
        if cwd:
            payload["cwd"] = cwd
        return self._request_json("POST", "/desktop/fork_thread", payload)

    def open_desktop_project(self, project_id: str) -> dict:
        return self._request_json("POST", "/desktop/open_project", {"project_id": project_id})

    def create_desktop_project(self, cwd: str) -> dict:
        return self._request_json("POST", "/desktop/new_project", {"cwd": cwd})

    def rename_desktop_project(self, cwd: str, title: str) -> dict:
        return self._request_json("POST", "/desktop/rename_project", {"cwd": cwd, "title": title})

    def remove_desktop_project(self, cwd: str) -> dict:
        return self._request_json("POST", "/desktop/remove_project", {"cwd": cwd})

    def interrupt_desktop_turn(
        self,
        thread_id: str | None = None,
        request_id: str | None = None,
        async_mode: bool = False,
    ) -> dict:
        payload: dict[str, str | bool] = {}
        if thread_id:
            payload["thread_id"] = thread_id
        if request_id:
            payload["request_id"] = request_id
        if async_mode:
            payload["async"] = True
        return self._request_json("POST", "/desktop/interrupt", payload)

    def rollback_desktop_turns(self, thread_id: str | None = None, num_turns: int = 1, async_mode: bool = False) -> dict:
        payload: dict[str, str | int | bool] = {"num_turns": max(1, int(num_turns or 1))}
        if thread_id:
            payload["thread_id"] = thread_id
        if async_mode:
            payload["async"] = True
        return self._request_json("POST", "/desktop/rollback", payload)

    def inject_desktop_items(self, items: list[dict], thread_id: str | None = None) -> dict:
        payload: dict[str, object] = {"items": items}
        if thread_id:
            payload["thread_id"] = thread_id
        return self._request_json("POST", "/desktop/inject_items", payload)

    def request_desktop_computer_control(self, instruction: str | None = None, thread_id: str | None = None) -> dict:
        payload: dict[str, str] = {}
        if instruction:
            payload["instruction"] = instruction
        if thread_id:
            payload["thread_id"] = thread_id
        return self._request_json("POST", "/desktop/computer_control", payload)

    def capture_desktop_screenshot(self) -> dict:
        return self._request_json("POST", "/desktop/screenshot", {})

    def build_ios_adhoc(
        self,
        thread_id: str | None = None,
        project_id: str | None = None,
        cwd: str | None = None,
        scheme: str | None = None,
        async_mode: bool = False,
    ) -> dict:
        payload: dict[str, str | bool] = {}
        if thread_id:
            payload["thread_id"] = thread_id
        if project_id:
            payload["project_id"] = project_id
        if cwd:
            payload["cwd"] = cwd
        if scheme:
            payload["scheme"] = scheme
        if async_mode:
            payload["async"] = True
        return self._request_json("POST", "/desktop/build_ios_adhoc", payload, timeout_seconds=2400)

    def get_ios_adhoc_build_status(self, request_id: str) -> dict:
        return self._request_json("GET", f"/desktop/build_ios_adhoc_status?request_id={quote(request_id)}")

    def restart_desktop_codex(self, thread_id: str | None = None) -> dict:
        payload: dict[str, str] = {}
        if thread_id:
            payload["thread_id"] = thread_id
        return self._request_json("POST", "/desktop/restart_codex", payload)

    def restart_desktop_connection(self) -> dict:
        return self._request_json("POST", "/desktop/restart_desktop_connection", {})

    def restart_mac(self) -> dict:
        return self._request_json("POST", "/desktop/restart_mac", {})

    def analyze_desktop_attachment(
        self,
        *,
        filename: str,
        mime_type: str,
        file_base64: str,
        thread_id: str | None = None,
    ) -> dict:
        payload = {
            "filename": filename,
            "mime_type": mime_type,
            "file_base64": file_base64,
        }
        if thread_id:
            payload["thread_id"] = thread_id
        return self._request_json("POST", "/desktop/attachment", payload)

    def upload_desktop_attachment(
        self,
        *,
        filename: str,
        mime_type: str,
        file_base64: str,
        thread_id: str | None = None,
    ) -> dict:
        payload = {
            "filename": filename,
            "mime_type": mime_type,
            "file_base64": file_base64,
        }
        if thread_id:
            payload["thread_id"] = thread_id
        return self._request_json("POST", "/desktop/upload_attachment", payload)

    def send_desktop_bridge_message(
        self,
        text: str,
        author: str = "mobile",
        mirrored_to_project: bool = False,
        thread_id: str | None = None,
        model: str | None = None,
        effort: str | None = None,
        permission: str | None = None,
        image_data_urls: list[str] | None = None,
        local_image_paths: list[str] | None = None,
        async_mode: bool = False,
    ) -> dict:
        payload = {"text": text, "author": author, "mirrored_to_project": mirrored_to_project}
        if thread_id:
            payload["thread_id"] = thread_id
        if model:
            payload["model"] = model
        if effort:
            payload["effort"] = effort
        if permission:
            payload["permission"] = permission
        if image_data_urls:
            payload["image_data_urls"] = image_data_urls
        if local_image_paths:
            payload["local_image_paths"] = local_image_paths
        if async_mode:
            payload["async"] = True
        return self._request_json(
            "POST",
            "/desktop/inbox",
            payload,
        )

    def reply_desktop_bridge_message(self, text: str) -> dict:
        return self._request_json("POST", "/desktop/reply", {"text": text})

    def mark_desktop_bridge_message_done(self, message_id: str) -> dict:
        return self._request_json("POST", "/desktop/mark_done", {"message_id": message_id})

    def _state_post(self, path: str, payload: dict) -> TaskState:
        return TaskState.model_validate(self._request_json("POST", path, payload)["state"])

    def _request_json(
        self,
        method: str,
        path: str,
        payload: dict | None = None,
        *,
        timeout_seconds: float | None = None,
    ) -> dict:
        body = None
        headers = {
            "Authorization": f"Bearer {self.token}",
            "Cache-Control": "no-cache",
            "Pragma": "no-cache",
        }
        if payload is not None:
            body = json.dumps(payload).encode("utf-8")
            headers["Content-Type"] = "application/json"
        req = request.Request(f"{self.base_url}{path}", data=body, headers=headers, method=method)
        retries = 2 if self._is_retryable_request(method, path) else 0
        last_error: Exception | None = None
        for attempt in range(retries + 1):
            try:
                with request.urlopen(req, timeout=timeout_seconds or self.timeout_seconds) as response:
                    return json.loads(response.read().decode("utf-8"))
            except error.HTTPError as exc:
                detail = exc.read().decode("utf-8", errors="replace")
                if attempt < retries and exc.code in {408, 429, 500, 502, 503, 504}:
                    time.sleep(0.8 * (attempt + 1))
                    continue
                parsed_detail = None
                try:
                    parsed = json.loads(detail)
                    if isinstance(parsed, dict):
                        parsed_detail = parsed
                except json.JSONDecodeError:
                    parsed_detail = None
                raise MacServiceHTTPError(exc.code, detail, parsed_detail) from exc
            except error.URLError as exc:
                last_error = exc
                if attempt < retries:
                    time.sleep(0.8 * (attempt + 1))
                    continue
                raise MacServiceError(f"Mac service unavailable: {exc.reason}") from exc
        if last_error is not None:
            raise MacServiceError(str(last_error))
        raise MacServiceError("Mac service unavailable")

    def _is_retryable_request(self, method: str, path: str) -> bool:
        if method.upper() == "GET":
            return True
        retryable_posts = {
            "/desktop/open_project",
            "/desktop/interrupt",
            "/desktop/rename_thread",
            "/desktop/archive_thread",
            "/desktop/unarchive_thread",
            "/desktop/pin_thread",
            "/desktop/unpin_thread",
            "/desktop/fork_thread",
            "/desktop/rename_project",
            "/desktop/remove_project",
            "/desktop/rollback",
            "/desktop/computer_control",
            "/desktop/screenshot",
            "/desktop/restart_codex",
            "/desktop/restart_desktop_connection",
            "/desktop/approve",
            "/desktop/reject",
        }
        normalized_path = path.split("?", 1)[0]
        return method.upper() == "POST" and normalized_path in retryable_posts

    def _cache_busted(self, path: str) -> str:
        separator = "&" if "?" in path else "?"
        return f"{path}{separator}_={int(time.time() * 1000)}"
