# Copyright 2026 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import asyncio
import json
import logging
from typing import Any, Protocol

from .dap_types import DapBaseModel
from .models import (
    AttachRequestArguments,
    ContinueArguments,
    ContinueResponse,
    DisconnectArguments,
    EvaluateArguments,
    EvaluateResponse,
    InitializeArguments,
    LaunchArguments,
    MessageType,
    PauseArguments,
    Response,
    ScopesArguments,
    ScopesResponse,
    SetBreakpointsArguments,
    SetBreakpointsResponse,
    StackTraceArguments,
    StackTraceResponse,
    StepOutArguments,
    ThreadsResponse,
    VariablesArguments,
    VariablesResponse,
)


class StreamWriterProtocol(Protocol):
    """Protocol for writer objects handling raw byte writes."""

    def write(self, data: bytes) -> None:
        ...

    async def drain(self) -> None:
        ...


READER_STOPPED_EVENT: str = "reader_stopped"
"""Synthetic event type emitted to `event_queue` when the reader task terminates.

This event is pushed to `event_queue` when the background reader task stops (due to EOF on stream,
cancellation, or an unrecoverable stream error).

* **DapClient State**: The `DapClient` instance is in the process of shutting down or already closed.
  `close()` is called immediately after emitting this event, cleaning up background reader/writer tasks
  and clearing pending requests. `is_running` will return `False`.
* **Subsequent Events**: This is the final event placed on `event_queue`. Callers must not expect any further
  events or responses from the client after receiving this.
* **Caller Expectations & Cleanup**: Callers consuming `event_queue` should catch this event and stop reading
  from the queue. The event payload is `{"type": READER_STOPPED_EVENT, "exception": exc}`, where `exception` is
  `None` if the reader exited cleanly or via cancellation, or the exception instance if caused by a failure.
  Callers should perform any necessary caller-side cleanup (e.g. tearing down test fixtures or session state).
  Once the client has finished closing (`is_running` is `False`), `run()` may be called again with a new reader, writer, and event_queue.
"""
logger = logging.getLogger(__name__)


class DapError(Exception):
    """Base exception for DAP client errors."""


class DapClient:
    """A client for the Debug Adapter Protocol."""

    DEFAULT_REQUEST_TIMEOUT: float = 5.0

    def __init__(self) -> None:
        """Initializes the DAP client."""
        self._pending_requests: dict[int, asyncio.Future[Any]] = {}
        self._seq_counter = 1
        self._write_queue: asyncio.Queue[
            tuple[
                dict[str, Any],
                asyncio.Future[None],
                asyncio.Future[Any],
            ]
        ] = asyncio.Queue()
        self._reader_task: asyncio.Task[None] | None = None
        self._writer_task: asyncio.Task[None] | None = None

    @property
    def is_running(self) -> bool:
        """Returns True if both the reader loop and writer tasks are currently running."""
        return (
            self._reader_task is not None
            and not self._reader_task.done()
            and self._writer_task is not None
            and not self._writer_task.done()
        )

    async def close(self) -> None:
        """Closes the client and cancels active writer and reader tasks if running."""
        curr_task = asyncio.current_task()
        tasks_to_cancel: list[asyncio.Task[None]] = []
        if (
            self._writer_task is not None
            and not self._writer_task.done()
            and self._writer_task != curr_task
        ):
            tasks_to_cancel.append(self._writer_task)
        self._writer_task = None

        if (
            self._reader_task is not None
            and not self._reader_task.done()
            and self._reader_task != curr_task
        ):
            tasks_to_cancel.append(self._reader_task)
        self._reader_task = None

        for task in tasks_to_cancel:
            task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

        while True:
            try:
                _, sent_fut, data_fut = self._write_queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            try:
                if not sent_fut.done():
                    sent_fut.cancel()
                if not data_fut.done():
                    data_fut.cancel()
            except Exception:
                pass
            finally:
                try:
                    self._write_queue.task_done()
                except ValueError:
                    pass

        for fut in self._pending_requests.values():
            if not fut.done():
                fut.cancel()
        self._pending_requests.clear()

    # The caller shouldn't directly manage this task. Use `run` and `close` instead.
    async def _run_writer_task(self, writer: StreamWriterProtocol) -> None:
        """Processes queued write requests sequentially in FIFO order.

        Args:
            writer: Output stream writer to transmit messages to debug adapter.
        """
        try:
            while True:
                # Cancellation is raised via client.close().
                request, sent_fut, data_fut = await self._write_queue.get()
                try:
                    # If the caller gave up or cancelled before dequeuing, skip sending.
                    if not data_fut.done():
                        await self._write_message(writer, request)
                        if not sent_fut.done():
                            sent_fut.set_result(None)
                    elif not sent_fut.done():
                        sent_fut.cancel()
                except Exception as e:
                    seq = request.get("seq")
                    if seq is not None:
                        self._pending_requests.pop(seq, None)
                    if not sent_fut.done():
                        sent_fut.set_exception(e)
                    if not data_fut.done():
                        data_fut.set_exception(e)
                    logger.error("DAP write message failed: %s", e)
                finally:
                    self._write_queue.task_done()
        except asyncio.CancelledError:
            pass

    async def _run_reader_task(
        self, reader: asyncio.StreamReader, event_queue: asyncio.Queue[Any]
    ) -> None:
        """Runs the client's read loop, processing messages from the reader.

        Args:
            reader: Input stream reader receiving messages from debug adapter.
            event_queue: Destination queue for incoming DAP events.
        """
        exc: Exception | None = None
        try:
            while True:
                msg = await self._read_message(reader)
                if msg is None:
                    break  # EOF

                msg_type = msg.get("type")
                if msg_type == MessageType.EVENT.value:
                    await event_queue.put(msg)
                elif msg_type == MessageType.RESPONSE.value:
                    req_seq = msg.get("request_seq")
                    if req_seq in self._pending_requests:
                        fut = self._pending_requests.pop(req_seq)
                        if not fut.done():
                            fut.set_result(msg)
        except asyncio.CancelledError:
            pass

        # We cannot recover when _read_message encounters errors such as an invalid Content-Length, as there is no reliable way to parse corrupt stream data.
        except Exception as e:
            logger.exception("Error in DAP client reader task")
            exc = e
            # Set exception on all in-flight pending request futures so callers fail fast with root cause.
            for fut in self._pending_requests.values():
                if not fut.done():
                    fut.set_exception(e)
            self._pending_requests.clear()
        finally:
            # Emit synthetic termination event so queue consumers wake up immediately and fail active event expectations without polling.
            try:
                await event_queue.put(
                    {"type": READER_STOPPED_EVENT, "exception": exc}
                )
            except Exception as put_err:
                logger.warning(
                    "Failed to emit READER_STOPPED_EVENT: %s", put_err
                )
            finally:
                await self.close()

    def run(
        self,
        reader: asyncio.StreamReader,
        writer: StreamWriterProtocol,
        event_queue: asyncio.Queue[Any],
    ) -> None:
        """Starts the client's read and write background tasks.

        Args:
            reader: Stream reader to receive messages from the debug adapter.
            writer: Stream writer to send messages to the debug adapter.
            event_queue: Queue to put received DAP events into.

        Raises:
            DapError: If the client is already running.
        """
        if self.is_running:
            raise DapError("DapClient is already running")

        self._reader_task = asyncio.create_task(
            self._run_reader_task(reader, event_queue)
        )
        self._writer_task = asyncio.create_task(self._run_writer_task(writer))

    # To make requests execute `_write_message` by FIFO order, we rely on the run_writer_task.
    def _send_request_future(
        self,
        command: str,
        arguments: DapBaseModel | None = None,
    ) -> tuple[int, asyncio.Future[None], asyncio.Future[dict[str, Any]]]:
        """Sends a request to the debug adapter synchronously by queueing it for the write worker.

        Args:
            command: The DAP command name.
            arguments: Optional arguments for the command.

        Returns:
            A tuple of (sequence number, sent future, response future).

        Raises:
            DapError: If the client is not running.
            TypeError: If arguments is not a DapBaseModel instance.
        """
        if not self.is_running:
            raise DapError(
                "DapClient is not running. Call 'run()' before sending requests."
            )

        seq = self._seq_counter
        self._seq_counter += 1

        loop = asyncio.get_running_loop()
        sent_fut: asyncio.Future[None] = loop.create_future()
        data_fut: asyncio.Future[dict[str, Any]] = loop.create_future()

        request: dict[str, Any] = {
            "seq": seq,
            "type": MessageType.REQUEST.value,
            "command": command,
        }
        if arguments is not None:
            if not isinstance(arguments, DapBaseModel):
                raise TypeError(
                    f"arguments must be a DapBaseModel, got {type(arguments)}"
                )
            request["arguments"] = arguments.dump_dap()

        self._pending_requests[seq] = data_fut
        self._write_queue.put_nowait((request, sent_fut, data_fut))

        return seq, sent_fut, data_fut

    async def _await_request_response(
        self,
        command: str,
        seq: int,
        sent_fut: asyncio.Future[None],
        data_fut: asyncio.Future[dict[str, Any]],
        timeout: float,
    ) -> dict[str, Any]:
        """Awaits wire transmission and response for a request with centralized cleanup."""
        try:
            async with asyncio.timeout(timeout):
                await sent_fut
                return await data_fut
        except TimeoutError:
            if not sent_fut.done():
                raise DapError(
                    f"Request {command} (seq={seq}) failed to write to socket within {timeout}s"
                )
            raise DapError(
                f"Request {command} (seq={seq}) sent over wire, but adapter timed out waiting for response after {timeout}s"
            )
        finally:
            self._pending_requests.pop(seq, None)
            if not sent_fut.done():
                sent_fut.cancel()
            if not data_fut.done():
                data_fut.cancel()

    async def _send_request(
        self,
        command: str,
        arguments: DapBaseModel | None = None,
        timeout: float = DEFAULT_REQUEST_TIMEOUT,
    ) -> dict[str, Any]:
        """Sends a request to the debug adapter and waits for the response.

        Args:
            command: The DAP command name.
            arguments: Optional arguments for the command.
            timeout: Maximum time to wait for response in seconds.

        Returns:
            The response message dictionary from the adapter.

        Raises:
            DapError: If the request times out or framing fails.
            ValueError: If timeout is not positive.
        """
        if timeout <= 0:
            raise ValueError(f"timeout must be positive, got {timeout}")

        # Instead of directly calling _write_message here,
        # we leverage _send_request_future, so this function inherently sends requests in FIFO order.
        # FIFO may not be necessary for this function and its client, but we keep it for consistency.
        seq, sent_fut, data_fut = self._send_request_future(command, arguments)
        resp = await self._await_request_response(
            command, seq, sent_fut, data_fut, timeout
        )
        if not resp.get("success", True):
            msg = resp.get("message", "Unknown DAP error")
            logger.error(
                "DAP request %s (seq=%s) failed: %s", command, seq, msg
            )
            raise DapError(f"DAP request {command} failed: {msg}")
        return resp

    async def initialize(self, args: InitializeArguments) -> Response:
        """Sends an initialize request.

        Args:
            args: Arguments for the initialize request.

        Returns:
            The response model.
        """
        resp = await self._send_request("initialize", args)
        return Response.model_validate(resp)

    async def disconnect(self, args: DisconnectArguments) -> Response:
        """Sends a disconnect request.

        Args:
            args: Arguments for the disconnect request.

        Returns:
            The response model.
        """
        resp = await self._send_request("disconnect", args)
        return Response.model_validate(resp)

    async def stack_trace(
        self, args: StackTraceArguments
    ) -> StackTraceResponse:
        """Sends a stackTrace request.

        Args:
            args: Arguments for the stackTrace request.

        Returns:
            The stackTrace response model.
        """
        resp = await self._send_request("stackTrace", args)
        return StackTraceResponse.model_validate(resp)

    async def continue_thread(
        self, args: ContinueArguments
    ) -> ContinueResponse:
        """Sends a continue request.

        Args:
            args: Arguments for the continue request.

        Returns:
            The continue response model.
        """
        resp = await self._send_request("continue", args)
        return ContinueResponse.model_validate(resp)

    async def pause_thread(self, args: PauseArguments) -> Response:
        """Sends a pause request.

        Args:
            args: Arguments for the pause request.

        Returns:
            The response model.
        """
        resp = await self._send_request("pause", args)
        return Response.model_validate(resp)

    async def step_out(self, args: StepOutArguments) -> Response:
        """Sends a stepOut request.

        Args:
            args: Arguments for the stepOut request.

        Returns:
            The response model.
        """
        resp = await self._send_request("stepOut", args)
        return Response.model_validate(resp)

    async def threads(self) -> ThreadsResponse:
        """Sends a threads request.

        Returns:
            The threads response model.
        """
        resp = await self._send_request("threads")
        return ThreadsResponse.model_validate(resp)

    async def attach(self, args: AttachRequestArguments) -> Response:
        """Sends an attach request.

        Args:
            args: Arguments for the attach request.

        Returns:
            The response model.
        """
        resp = await self._send_request("attach", args)
        return Response.model_validate(resp)

    async def launch(self, args: LaunchArguments) -> Response:
        """Sends a launch request.

        Args:
            args: Arguments for the launch request.

        Returns:
            The response model.
        """
        resp = await self._send_request("launch", args)
        return Response.model_validate(resp)

    async def evaluate(self, args: EvaluateArguments) -> EvaluateResponse:
        """Sends an evaluate request.

        Args:
            args: Arguments for the evaluate request.

        Returns:
            The response model.
        """
        resp = await self._send_request("evaluate", args)
        return EvaluateResponse.model_validate(resp)

    async def scopes(self, args: ScopesArguments) -> ScopesResponse:
        """Sends a scopes request.

        Args:
            args: Arguments for the scopes request.

        Returns:
            The scopes response model.
        """
        resp = await self._send_request("scopes", args)
        return ScopesResponse.model_validate(resp)

    async def variables(self, args: VariablesArguments) -> VariablesResponse:
        """Sends a variables request.

        Args:
            args: Arguments for the variables request.

        Returns:
            The variables response model.
        """
        resp = await self._send_request("variables", args)
        return VariablesResponse.model_validate(resp)

    async def set_breakpoints(
        self, args: SetBreakpointsArguments
    ) -> SetBreakpointsResponse:
        """Sends a setBreakpoints request.

        Args:
            args: Arguments for the setBreakpoints request.

        Returns:
            The setBreakpoints response model.
        """
        resp = await self._send_request("setBreakpoints", args)
        return SetBreakpointsResponse.model_validate(resp)

    async def _read_message(
        self, reader: asyncio.StreamReader
    ) -> dict[str, Any] | None:
        """Reads a single message from the reader, handling protocol framing.

        Args:
            reader: Stream reader to read from.

        Returns:
            The parsed message dictionary, or None on EOF.

        Raises:
            DapError: If framing headers are invalid or missing.
        """
        content_length = None
        while True:
            line = await reader.readline()
            if not line:
                return None  # EOF
            trimmed = line.decode("utf-8").strip()
            if not trimmed:
                break  # End of headers

            if trimmed.startswith("Content-Length:"):
                parts = trimmed.split(":")
                if len(parts) >= 2:
                    try:
                        content_length = int(parts[1].strip())
                    except ValueError:
                        raise DapError(f"Invalid Content-Length: {trimmed}")

        if content_length is None:
            raise DapError("Missing Content-Length header")

        body = await reader.readexactly(content_length)
        return json.loads(body.decode("utf-8"))

    async def _write_message(
        self, writer: StreamWriterProtocol, value: dict[str, Any]
    ) -> None:
        """Writes a message to the writer, handling protocol framing.

        Args:
            writer: Stream writer to write to.
            value: The message dictionary to serialize and send.
        """
        content = json.dumps(value, separators=(",", ":")).encode("utf-8")
        header = f"Content-Length: {len(content)}\r\n\r\n".encode("utf-8")
        writer.write(header)
        writer.write(content)
        await writer.drain()