import json
import logging
import threading
import time
import uuid
import weakref
from collections import OrderedDict
from http.server import ThreadingHTTPServer
from queue import SimpleQueue
from typing import Any, Callable, Generator, Optional, Union, cast

from core.runtime.BackgroundRuntime import BackgroundRuntime
from httpservice.HttpMetrics import HttpMetrics
from httpservice.HttpServerBaseMixin import HttpServerBaseMixin
from httpservice.McpMixin import McpMixin
from httpservice.handlers.HttpRequestHandler import HttpRequestHandler, StreamingResponseData


class BackgroundHttpServer(HttpServerBaseMixin, McpMixin):
    MAX_RESULTS_CACHE: int = 1000

    def __init__(self, runtime: BackgroundRuntime, port: int, timeout: int, token: Optional[str] = None,
                 metrics_enabled: bool = True,
                 metrics_slow_threshold_ms: int = 5000,
                 metrics_log_interval_seconds: int = 60,
                 metrics_quantile_window: int = 256) -> None:
        self.runtime = runtime
        self._port = int(port)
        self._timeout = int(timeout)
        self._token = token
        self._host = ""
        self._httpd: Optional[ThreadingHTTPServer] = None
        self._running = False

        self._result_store: OrderedDict[str, Any] = OrderedDict()
        self._result_events: weakref.WeakValueDictionary[str, threading.Event] = weakref.WeakValueDictionary()
        self._result_lock: threading.RLock = threading.RLock()

        # Metrics: in-memory, BG-only. GUI server doesn't attach _metrics so the
        # handler hook short-circuits there.
        self._metrics: Optional[HttpMetrics] = (
            HttpMetrics(
                slow_threshold_ms=int(metrics_slow_threshold_ms),
                quantile_window=int(metrics_quantile_window),
                executor_provider=lambda: getattr(self, "_executor", None),
            ) if metrics_enabled else None
        )
        self._metrics_log_interval: int = int(metrics_log_interval_seconds)

        HttpRequestHandler.register_view(None)
        HttpRequestHandler.register_server(self)
        HttpRequestHandler.register_routes(self._build_route_map())

    def _build_route_map(self) -> dict[str, Callable]:
        return {
            "GET:/": self._handle_index,
            "GET:/health": self._handle_health,
            "GET:/petp/tools": self._handle_petp_tools,
            "POST:/petp/exec": self._handle_petp_exec,
            "GET:/petp/result": self._handle_result_check,
            "GET:/petp/crons": self._handle_cron_list,
            "GET:/petp/crons/history": self._handle_cron_history,
            "GET:/petp/metrics": self._handle_metrics,
            "GET:/mcp": self._handle_mcp_get,
            "POST:/mcp": self._handle_mcp,
            "DELETE:/mcp": self._handle_mcp,
        }

    def _handle_index(self, handler: HttpRequestHandler, params: Optional[dict] = None) -> dict:
        return {
            "server": "PETP Background HTTP Server",
            "available_endpoints": ["/health", "/petp/exec", "/petp/tools", "/petp/result", "/petp/crons",
                                    "/petp/crons/history", "/petp/metrics", "/mcp"],
        }

    def _handle_petp_tools(self, handler: HttpRequestHandler, payload: Optional[dict] = None) -> Union[dict, tuple]:
        err = self._require_token(handler)
        if err is not None:
            return err
        return self.runtime.get_tools()

    def _handle_petp_exec(self, handler: HttpRequestHandler, payload: Optional[dict]) -> Union[dict, tuple, StreamingResponseData]:
        err = self._require_token(handler)
        if err is not None:
            return err
        if not payload or "action" not in payload or "params" not in payload:
            return {"error": "Missing required 'action' or 'params' parameter"}, 400

        action = payload.get("action", "execution")
        params = payload.get("params", {})
        if not isinstance(params, dict):
            return {"error": "params must be a JSON object"}, 400

        if self._wants_sse(handler):
            return self._stream_exec_result(action, params)

        wait_for_result = payload.get("wait_for_result", True)
        if not isinstance(wait_for_result, bool):
            wait_for_result = str(wait_for_result).lower() == "true"

        request_id = self._generate_request_id()

        if action == "execution":
            execution_name = params.get("execution")
            if not isinstance(execution_name, str) or not execution_name:
                return {"error": "params.execution is required"}, 400
            exec_name: str = cast(str, execution_name)

            if wait_for_result:
                runtime_result = self.runtime.run_execution(exec_name, params)
                client_result = self._extract_business_response(runtime_result)
                return client_result, 200 if bool(runtime_result.get("ok")) else 500

            self._submit_async(request_id, lambda: self.runtime.run_execution(exec_name, params))
            return {
                "status": "pending",
                "request_id": request_id,
                "message": "Request is being processed. Use GET /petp/result?request_id=<id> to check status.",
            }, 202

        if action == "pipeline":
            pipeline_name = params.get("pipeline")
            if not isinstance(pipeline_name, str) or not pipeline_name:
                return {"error": "params.pipeline is required"}, 400
            pipe_name: str = cast(str, pipeline_name)

            if wait_for_result:
                runtime_result = self.runtime.run_pipeline(pipe_name, params)
                client_result = self._extract_business_response(runtime_result)
                return client_result, 200 if bool(runtime_result.get("ok")) else 500

            self._submit_async(request_id, lambda: self.runtime.run_pipeline(pipe_name, params))
            return {
                "status": "pending",
                "request_id": request_id,
                "message": "Request is being processed. Use GET /petp/result?request_id=<id> to check status.",
            }, 202

        return {"error": f"Unsupported action: {action}"}, 400

    _SSE_HEARTBEAT_INTERVAL: float = 1.0

    def _stream_exec_result(self, action: str, params: dict) -> StreamingResponseData:
        progress_queue: SimpleQueue = SimpleQueue()
        cancel_event = threading.Event()
        result_holder: dict = {}

        if action == "execution":
            execution_name = params.get("execution")
            if not isinstance(execution_name, str) or not execution_name:
                return self._sse_immediate_error("params.execution is required")
            exec_name: str = cast(str, execution_name)

            def runner():
                try:
                    result_holder["r"] = self.runtime.run_execution(
                        exec_name, params, progress_queue=progress_queue
                    )
                except Exception as e:
                    result_holder["r"] = {"ok": False, "error": str(e),
                                          "data": {}, "meta": {}}
        elif action == "pipeline":
            pipeline_name = params.get("pipeline")
            if not isinstance(pipeline_name, str) or not pipeline_name:
                return self._sse_immediate_error("params.pipeline is required")
            pipe_name: str = cast(str, pipeline_name)

            def runner():
                try:
                    result_holder["r"] = self.runtime.run_pipeline(
                        pipe_name, params, progress_queue=progress_queue
                    )
                except Exception as e:
                    result_holder["r"] = {"ok": False, "error": str(e),
                                          "data": {}, "meta": {}}
        else:
            return self._sse_immediate_error(f"Unsupported action: {action}")

        def generator() -> Generator[str, None, None]:
            future = self._get_executor().submit(runner)
            while not future.done():
                if cancel_event.is_set():
                    return
                drained = self._drain_progress_for_petp(progress_queue)
                if drained:
                    for evt in drained:
                        if cancel_event.is_set():
                            return
                        yield evt
                else:
                    yield self._sse_progress_event({"type": "heartbeat", "ts": int(time.time())})
                try:
                    future.result(timeout=self._SSE_HEARTBEAT_INTERVAL)
                except Exception:
                    pass
            for evt in self._drain_progress_for_petp(progress_queue):
                if cancel_event.is_set():
                    return
                yield evt
            if cancel_event.is_set():
                return
            result = result_holder.get("r", {"ok": False, "error": "no result", "data": {}, "meta": {}})
            client_result = self._extract_business_response(result)
            yield self._sse_result_event(client_result)

        return StreamingResponseData(
            generator(),
            {"Cache-Control": "no-cache"},
            "text/event-stream",
            200,
            cancel_event=cancel_event,
        )

    def _drain_progress_for_petp(self, queue: SimpleQueue) -> list[str]:
        events: list[str] = []
        while not queue.empty():
            try:
                msg = queue.get_nowait()
            except Exception:
                break
            if isinstance(msg, dict):
                events.append(self._sse_progress_event(msg))
            else:
                events.append(self._sse_progress_event({"type": "info", "message": str(msg)}))
        return events

    @staticmethod
    def _sse_progress_event(payload: dict) -> str:
        return f"event: progress\ndata: {json.dumps(payload, ensure_ascii=False, default=str)}\n\n"

    @staticmethod
    def _sse_result_event(payload: Any) -> str:
        return f"event: result\ndata: {json.dumps(payload, ensure_ascii=False, default=str)}\n\n"

    def _sse_immediate_error(self, msg: str) -> StreamingResponseData:
        def gen():
            yield self._sse_result_event({"ok": False, "error": msg, "data": {}, "meta": {}})
        return StreamingResponseData(gen(), {"Cache-Control": "no-cache"}, "text/event-stream", 200)

    def _handle_result_check(self, handler: HttpRequestHandler, params: Optional[dict] = None) -> Union[dict, tuple]:
        err = self._require_token(handler)
        if err is not None:
            return err
        if not params or "request_id" not in params:
            return {"error": "Missing request_id parameter"}, 400

        request_id = params["request_id"]
        result = self.get_result(request_id)
        if result is None:
            with self._result_lock:
                if request_id in self._result_events:
                    return {
                        "status": "pending",
                        "request_id": request_id,
                        "message": "Request is still being processed",
                    }, 200
            return {"error": "Request not found or expired"}, 404
        return result, self._status_code_for_result(result)

    def _handle_cron_list(self, handler: HttpRequestHandler, params: Optional[dict] = None) -> Union[dict, tuple]:
        err = self._require_token(handler)
        if err is not None:
            return err
        return {"crons": self.runtime.get_active_crons()}

    def _handle_cron_history(self, handler: HttpRequestHandler, params: Optional[dict] = None) -> Union[dict, tuple]:
        err = self._require_token(handler)
        if err is not None:
            return err
        params = params or {}
        pipeline_name = params.get("pipeline")
        limit = int(params.get("limit", 50))
        record_id = params.get("id")

        if record_id:
            record = self.runtime.get_cron_record(record_id)
            if record is None:
                return {"error": "Record not found"}, 404
            return record

        return {"history": self.runtime.get_cron_history(pipeline_name, limit)}

    def _handle_metrics(self, handler: HttpRequestHandler, params: Optional[dict] = None) -> Union[dict, tuple]:
        err = self._require_token(handler)
        if err is not None:
            return err
        if self._metrics is None:
            return {"error": "metrics disabled"}, 503
        return self._metrics.snapshot()

    def _handle_mcp(
            self,
            handler: HttpRequestHandler,
            params: Optional[dict] = None,
    ) -> Union[StreamingResponseData, tuple]:
        params = params or {}
        err = self._require_token(handler)
        if err is not None:
            return err

        # Batch request support
        if '_batch' in params:
            return self._handle_mcp_batch(handler, params['_batch'], self._handle_mcp)

        session_id = self._extract_session_id(handler)
        method = params.get("method")

        if not method:
            return self._mcp_method_not_found(params.get("id"), "(empty)", session_id, handler)

        if method == "initialize":
            session_id = session_id or (uuid.uuid4().hex + uuid.uuid4().hex)
            protocol_version = params.get("params", {}).get("protocolVersion") or "2025-11-25"
            return self._mcp_initialize_response(params.get("id"), protocol_version, session_id, "PETP Background MCP", handler)

        if method == "notifications/initialized":
            return self._mcp_initialized_response(session_id)

        if method == "tools/list":
            return self._mcp_tools_list_response(params.get("id"), session_id, self.runtime.get_tools(), handler)

        if method == "tools/call":
            request_id = params.get("id")
            tool_name = params.get("params", {}).get("name")
            arguments = params.get("params", {}).get("arguments", {})
            if not isinstance(tool_name, str) or not tool_name:
                return self._single_sse_response({
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {"code": -32602, "message": "Invalid params: params.name is required"},
                }, session_id)
            tool_exec_name: str = cast(str, tool_name)
            arguments_json = json.dumps(arguments, ensure_ascii=False)
            info = f"call tool: {tool_exec_name} with params: {arguments_json}"
            logging.info("MCP tools/call: %s", info)

            if not self._wants_sse(handler):
                raw_result = self._run_with_timeout(
                    lambda: self.runtime.run_execution(tool_exec_name, arguments),
                    self._timeout,
                )
                return self._mcp_tools_call_json_response(
                    request_id, session_id, raw_result, tool_exec_name,
                    self.runtime.get_tools, handler,
                )

            def _stream_call_result() -> Generator[str, None, None]:
                yield self._sse_event({
                    "jsonrpc": "2.0",
                    "method": "notifications/message",
                    "params": {"level": "info", "data": info},
                })

                progress_queue = SimpleQueue()
                result = None
                for item in self._run_with_progress(
                    lambda: self.runtime.run_execution(tool_exec_name, arguments, progress_queue=progress_queue),
                    self._timeout, tool_exec_name, progress_queue=progress_queue,
                    cancel_event=cancel_event,
                ):
                    if isinstance(item, str):
                        yield item
                    else:
                        result = item

                if cancel_event.is_set():
                    return

                client_result, structured_content = self._build_tools_call_result(
                    tool_exec_name, result, self.runtime.get_tools
                )

                is_error = self._is_mcp_error_result(result)
                if is_error:
                    error_msg = structured_content.get("error", "unknown error") if isinstance(structured_content, dict) else "unknown error"
                    content_text = f"{tool_exec_name}: error: {error_msg}"
                else:
                    content_text = f"{tool_exec_name}: {self._to_mcp_text(structured_content)}"
                if isinstance(result, dict) and result.get("meta") is not None:
                    logging.info("MCP tools/call meta for %s: %s", tool_name,
                                 json.dumps(result.get("meta"), ensure_ascii=False, default=str))
                yield self._sse_event({
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {
                        "content": [{"type": "text", "text": content_text}],
                        "structuredContent": structured_content,
                        "isError": is_error,
                    },
                })

            cancel_event = threading.Event()
            return StreamingResponseData(_stream_call_result(), self._build_sse_headers(session_id),
                                         "text/event-stream", 200, cancel_event=cancel_event)

        return self._mcp_method_not_found(params.get("id"), method, session_id, handler)

    def _submit_async(self, request_id: str, task: Callable[[], dict]) -> None:
        event = threading.Event()
        with self._result_lock:
            self._result_events[request_id] = event

        def _runner():
            runtime_result = task()
            client_result = self._extract_business_response(runtime_result)
            self.store_result(request_id, client_result)

        threading.Thread(target=_runner, daemon=True).start()

    def store_result(self, request_id: str, result: Any) -> None:
        with self._result_lock:
            if len(self._result_store) >= self.MAX_RESULTS_CACHE:
                oldest_key = next(iter(self._result_store))
                del self._result_store[oldest_key]
            self._result_store[request_id] = result
            event = self._result_events.get(request_id)
            if event and not event.is_set():
                event.set()

    def get_result(self, request_id: str, remove: bool = True) -> Any:
        with self._result_lock:
            if request_id not in self._result_store:
                return None
            result = self._result_store[request_id]
            if remove:
                del self._result_store[request_id]
            return result

    def start(self) -> None:
        if self._running:
            return
        handler_class = cast(Any, HttpRequestHandler)
        self._httpd = ThreadingHTTPServer((self._host, self._port), handler_class)
        threading.Thread(target=self._httpd.serve_forever, daemon=True).start()
        self._running = True
        logging.info("PETP Background HTTP Server is serving at http://%s:%d", self._host or "localhost", self._port)
        threading.Thread(target=self.runtime.warm_processor_cache, daemon=True).start()
        if self._metrics is not None and self._metrics_log_interval > 0:
            threading.Thread(target=self._metrics_log_loop, daemon=True).start()

    def _metrics_log_loop(self) -> None:
        """Periodically dump a metrics snapshot to log with the ``METRICS `` prefix."""
        while self._running:
            time.sleep(self._metrics_log_interval)
            if not self._running:
                break
            try:
                snap = self._metrics.snapshot()
                logging.info("METRICS %s", json.dumps(snap, ensure_ascii=False, default=str))
            except Exception:
                logging.exception("metrics log loop error")

    def stop(self) -> None:
        if self._httpd and self._running:
            self._httpd.shutdown()
            self._shutdown_executor()
            self._running = False
            logging.info("PETP Background HTTP Server has been stopped")
