"""注册表 CRUD — 纯元数据管理，读写 registry.json"""

from __future__ import annotations

import asyncio
import json
import logging
import shutil
import time
import threading
from pathlib import Path

from tool_agent.config import settings
from tool_agent.models import ToolMeta, ToolStatus, BackendRuntime

logger = logging.getLogger(__name__)


class RegistryWriteQueue:
    """Registry 写入队列，保证并发安全"""

    def __init__(self):
        self._queue: asyncio.Queue = None
        self._worker_task = None
        self._running = False

    async def start(self):
        """启动写入队列"""
        if self._running:
            return
        self._queue = asyncio.Queue()
        self._running = True
        self._worker_task = asyncio.create_task(self._worker())
        logger.info("Registry write queue started")

    async def stop(self):
        """停止写入队列"""
        if not self._running:
            return
        self._running = False
        await self._queue.put(None)  # 发送停止信号
        if self._worker_task:
            await self._worker_task
        logger.info("Registry write queue stopped")

    async def enqueue(self, operation: str, data: dict):
        """将写入操作加入队列"""
        if not self._running:
            await self.start()
        await self._queue.put({"operation": operation, "data": data})

    async def _worker(self):
        """后台工作线程，顺序处理写入操作"""
        while self._running:
            try:
                item = await self._queue.get()
                if item is None:  # 停止信号
                    break

                operation = item["operation"]
                data = item["data"]

                # 执行写入操作
                if operation == "save":
                    callback = data.get("callback")
                    if callback:
                        callback()

                self._queue.task_done()
            except Exception as e:
                logger.error(f"Registry write queue error: {e}")


class ToolRegistry:
    """工具注册表 — 只管工具是什么，不管在哪跑"""

    def __init__(self) -> None:
        self._path = settings.data_dir / "registry.json"
        self._tools: dict[str, ToolMeta] = {}
        self._lock = threading.Lock()
        self._write_queue = RegistryWriteQueue()

        # 工具组管理器
        from tool_agent.registry.groups import ToolGroupManager
        self.group_manager = ToolGroupManager()

        self._load()

    def _load(self) -> None:
        with self._lock:
            self._tools = {}
            if self._path.exists():
                data = json.loads(self._path.read_text(encoding="utf-8"))
                for item in data.get("tools", []):
                    tool = ToolMeta(**item)
                    if tool.backend_runtime is None:
                        tool.backend_runtime = self._infer_backend_runtime(tool.tool_id)
                    if not tool.group_ids:
                        tool.group_ids = [g.group_id for g in self.group_manager.find_groups_for_tool(tool.tool_id)]
                    self._tools[tool.tool_id] = tool

    def _save(self) -> None:
        """同步保存（内部使用，已加锁）"""
        self._path.parent.mkdir(parents=True, exist_ok=True)
        data = {"tools": [t.model_dump(mode="json") for t in self._tools.values()], "version": "2.0"}
        self._path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")

    async def _save_async(self) -> None:
        """异步保存（通过队列）"""
        await self._write_queue.enqueue("save", {"callback": self._save_sync})

    def _save_sync(self) -> None:
        """同步保存回调"""
        with self._lock:
            self._save()

    def _infer_backend_runtime(self, tool_id: str) -> BackendRuntime:
        """根据 source 推断后端执行环境"""
        source_store_path = settings.data_dir / "sources.json"
        if source_store_path.exists():
            try:
                data = json.loads(source_store_path.read_text(encoding="utf-8"))
                for source in data.get("sources", {}).get(tool_id, []):
                    stype = source.get("type", "")
                    if stype == "docker":
                        return BackendRuntime.DOCKER
                    if stype == "local":
                        return BackendRuntime.LOCAL
                    if stype in {"hub", "remote"}:
                        return BackendRuntime.REMOTE
            except Exception:
                pass
        return BackendRuntime.LOCAL

    # ---- CRUD ----

    def get(self, tool_id: str) -> ToolMeta | None:
        return self._tools.get(tool_id)

    def list_all(self) -> list[ToolMeta]:
        return list(self._tools.values())

    def register(self, tool: ToolMeta) -> None:
        with self._lock:
            self._tools[tool.tool_id] = tool
        # 异步保存，避免阻塞
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                asyncio.create_task(self._save_async())
            else:
                self._save()
        except RuntimeError:
            # 没有事件循环，同步保存
            with self._lock:
                self._save()
        logger.info(f"Registered tool: {tool.tool_id}")

    def unregister(self, tool_id: str) -> bool:
        with self._lock:
            if tool_id in self._tools:
                del self._tools[tool_id]
                self._save()
                logger.info(f"Unregistered tool: {tool_id}")
                return True
        return False

    def update(self, tool: ToolMeta) -> None:
        with self._lock:
            self._tools[tool.tool_id] = tool
        # 异步保存
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                asyncio.create_task(self._save_async())
            else:
                self._save()
        except RuntimeError:
            with self._lock:
                self._save()

    # ---- 查询 ----

    def list_active(self) -> list[ToolMeta]:
        return [t for t in self._tools.values() if t.status == ToolStatus.ACTIVE]

    def find_by_category(self, category: str) -> list[ToolMeta]:
        """按功能分类查找（保留向后兼容）"""
        return [t for t in self._tools.values() if t.category == category and t.status == ToolStatus.ACTIVE]

    def find_by_backend_runtime(self, backend_runtime: str) -> list[ToolMeta]:
        """按后端执行环境查找"""
        try:
            br = BackendRuntime(backend_runtime)
            return [t for t in self._tools.values() if t.backend_runtime == br and t.status == ToolStatus.ACTIVE]
        except ValueError:
            return []

    def get_tools_in_group(self, group_id: str) -> list[ToolMeta]:
        """获取工具组内的所有工具"""
        group = self.group_manager.get_group(group_id)
        if not group:
            return []
        return [self._tools[tid] for tid in group.tool_ids if tid in self._tools and self._tools[tid].status == ToolStatus.ACTIVE]

    def search(self, keyword: str) -> list[ToolMeta]:
        kw = keyword.lower()
        return [
            t for t in self._tools.values()
            if t.status == ToolStatus.ACTIVE
            and (kw in t.name.lower() or kw in t.description.lower())
        ]

    # ---- 清理 ----

    @staticmethod
    def _kill_processes_in_dir(directory: Path) -> None:
        try:
            import psutil
        except ImportError:
            return

        dir_str = str(directory.resolve())
        killed = []
        for proc in psutil.process_iter(["pid", "cwd", "open_files"]):
            try:
                if proc.info.get("cwd") and str(Path(proc.info["cwd"]).resolve()).startswith(dir_str):
                    logger.info(f"Killing process {proc.pid} (cwd in {directory.name})")
                    proc.kill()
                    killed.append(proc)
                    continue
                open_files = proc.info.get("open_files")
                if open_files:
                    for f in open_files:
                        if str(Path(f.path).resolve()).startswith(dir_str):
                            logger.info(f"Killing process {proc.pid} (open file in {directory.name})")
                            proc.kill()
                            killed.append(proc)
                            break
            except Exception:
                pass

        for proc in killed:
            try:
                proc.wait(timeout=5)
            except Exception:
                pass
        if killed:
            time.sleep(0.5)

    @staticmethod
    def _safe_rmtree(path: Path) -> bool:
        try:
            shutil.rmtree(path)
            return True
        except PermissionError:
            time.sleep(1)
            try:
                shutil.rmtree(path)
                return True
            except Exception as e:
                logger.warning(f"Failed to delete {path}: {e}")
                return False

    def destroy(self, tool_id: str) -> dict:
        """彻底删除工具：注册信息 + 项目文件 + staging 残留"""
        cleaned = []

        if self.unregister(tool_id):
            cleaned.append("registry")

        local_dir = settings.tools_dir / "local" / tool_id
        if local_dir.exists():
            self._kill_processes_in_dir(local_dir)
            if self._safe_rmtree(local_dir):
                cleaned.append(f"local:{local_dir}")
                logger.info(f"Deleted local dir: {local_dir}")

        staging_dir = settings.data_dir / "staging"
        if staging_dir.exists():
            for d in staging_dir.iterdir():
                if not d.is_dir():
                    continue
                try:
                    for f in d.rglob("*"):
                        if f.is_file() and tool_id in f.read_text(errors="ignore")[:1000]:
                            self._safe_rmtree(d)
                            cleaned.append(f"staging:{d.name}")
                            break
                except Exception:
                    pass

        # 清理来源存储
        try:
            from tool_agent.router.status import SourceStore
            SourceStore().remove_tool(tool_id)
            cleaned.append("sources")
        except Exception:
            pass

        logger.info(f"Destroyed tool '{tool_id}': {cleaned}")
        return {"status": "success", "tool_id": tool_id, "cleaned": cleaned}

    def destroy_all(self) -> dict:
        tool_ids = [t.tool_id for t in self.list_all()]
        results = []
        for tid in tool_ids:
            results.append(self.destroy(tid))

        local_base = settings.tools_dir / "local"
        if local_base.exists():
            for d in local_base.iterdir():
                if d.is_dir() and d.name not in tool_ids:
                    self._kill_processes_in_dir(d)
                    if self._safe_rmtree(d):
                        logger.info(f"Deleted orphan local dir: {d.name}")
                        results.append({"tool_id": d.name, "cleaned": [f"local:{d}"]})

        staging_dir = settings.data_dir / "staging"
        if staging_dir.exists():
            for d in staging_dir.iterdir():
                if d.is_dir():
                    self._safe_rmtree(d)
            logger.info("Cleared all staging directories")

        return {"status": "success", "count": len(results), "results": results}
