# This file is part of nbxmpp.
#
# SPDX-License-Identifier: GPL-3.0-or-later

from __future__ import annotations

from typing import Any
from typing import Literal

import logging
from collections.abc import Callable
from pathlib import Path

from gi.repository import Gio
from gi.repository import GLib
from gi.repository import GObject
from gi.repository import Soup

import nbxmpp
from nbxmpp.const import HTTPRequestError

log = logging.getLogger("nbxmpp.http")


HTTP_METHODS_T = Literal[
    "CONNECT",
    "DELETE",
    "GET",
    "HEAD",
    "OPTIONS",
    "PATCH",
    "POST",
    "PUT",
    "TRACE",
]
CHUNK_SIZE = 32768
DEFAULT_USER_AGENT = f"nbxmpp/{nbxmpp.__version__}"
SIGNAL_ACTIONS = GObject.SignalFlags.RUN_LAST | GObject.SignalFlags.ACTION
MIN_SOUP_3_3_0 = Soup.check_version(3, 3, 0)


class HTTPLogAdapter(logging.LoggerAdapter):
    def process(self, msg: str, kwargs: Any) -> tuple[str, Any]:
        return f'{self.extra["request"]}: {msg}', kwargs


class HTTPSession:
    def __init__(self, user_agent: str = DEFAULT_USER_AGENT) -> None:
        self._session = Soup.Session()
        self._session.set_user_agent(user_agent)
        self._session.add_feature_by_type(Soup.ContentSniffer)
        self._session.remove_feature_by_type(Soup.ContentDecoder)

    def get_soup_session(self) -> Soup.Session:
        return self._session

    def set_proxy_resolver(self, resolver: Gio.SimpleProxyResolver | None) -> None:

        self._session.set_proxy_resolver(resolver)

    def create_request(self, force_http1: bool = False) -> HTTPRequest:
        return HTTPRequest(self, force_http1=force_http1)


class HTTPRequest(GObject.GObject):

    __gtype_name__ = "HTTPRequest"

    __gsignals__ = {
        "content-sniffed": (SIGNAL_ACTIONS, None, (int, str)),
        "starting-response-body": (SIGNAL_ACTIONS, None, ()),
        "response-progress": (SIGNAL_ACTIONS, None, (float,)),
        "request-progress": (SIGNAL_ACTIONS, None, (float,)),
        "finished": (SIGNAL_ACTIONS, None, ()),
        "destroy": (SIGNAL_ACTIONS, None, ()),
    }

    def __init__(self, session: HTTPSession, force_http1: bool = False) -> None:
        GObject.GObject.__init__(self)

        self._log = HTTPLogAdapter(log, extra={"request": self})

        self._session = session

        self._received_size = 0
        self._sent_size = 0

        self._cancellable = Gio.Cancellable()
        self._input_stream: Gio.InputStream | None = None
        self._output_stream: Gio.OutputStream | None = None
        self._is_finished = False
        self._error: HTTPRequestError | None = None
        self._is_complete = False
        self._timeout_reached = False
        self._timeout_id: int | None = None
        self._accept_certificate_func = None

        self._response_body_file: Gio.File | None = None
        self._response_body_data = b""
        self._body_received = False

        self._request_body_file: Gio.File | None = None
        self._request_body_data: bytes | None = None

        self._request_content_type = ""
        self._request_content_length = 0

        self._response_content_type = ""
        self._response_content_length = 0

        self._emit_request_progress = False
        self._emit_response_progress = False

        self._message = Soup.Message()
        if MIN_SOUP_3_3_0:
            self._message.set_force_http1(force_http1)  # type: ignore
        self._user_data = None

        self._log.info("Created")

    def is_finished(self) -> bool:
        return self._is_finished

    def is_complete(self) -> bool:
        return self._is_complete

    def get_error(self) -> HTTPRequestError | None:
        return self._error

    def get_error_string(self) -> str:
        if self._error == HTTPRequestError.STATUS_NOT_OK:
            return self._message.get_reason_phrase()
        return repr(self._error)

    def get_response_headers(self) -> Soup.MessageHeaders:
        return self._message.get_response_headers()

    def get_request_headers(self) -> Soup.MessageHeaders:
        return self._message.get_request_headers()

    def get_data(self) -> bytes:
        if not self._is_finished:
            raise ValueError("Process not finished, data not available")
        return self._response_body_data

    def get_uri(self) -> GLib.Uri | None:
        return self._message.get_uri()

    def get_status(self) -> Soup.Status:
        return self._message.get_status()

    def get_soup_message(self) -> Soup.Message:
        return self._message

    def set_user_data(self, user_data: Any) -> None:
        self._user_data = user_data

    def get_user_data(self) -> Any:
        return self._user_data

    def cancel(self) -> None:
        if self._is_finished:
            self._log.warning("Cancel requested after download has already finished")
            return

        self._log.info("Cancel requested")
        self._cancellable.cancel()

    def set_accept_certificate_func(self, func: Any) -> None:
        self._accept_certificate_func = func

    def set_request_body_from_path(self, content_type: str, path: Path) -> None:
        if not path.exists():
            raise ValueError("%s does not exist" % path)

        if not path.is_file():
            raise ValueError("%s is not a file" % path)

        self._request_body_file = Gio.File.new_for_path(str(path))
        self._request_content_type = content_type
        self._request_content_length = path.stat().st_size

    def set_request_body(self, content_type: str, data: bytes) -> None:
        self._request_body_data = data
        self._request_content_type = content_type
        self._request_content_length = len(data)

    def set_response_body_from_path(self, path: Path) -> None:
        self._response_body_file = Gio.File.new_for_path(str(path))

    def connect(self, signal_name: str, callback: Any, *args: Any) -> None:

        if signal_name == "response-progress":
            self._emit_response_progress = True

        if signal_name == "request-progress":
            self._emit_request_progress = True

        if signal_name in GObject.signal_list_names(HTTPRequest):
            GObject.GObject.connect(self, signal_name, callback, *args)
            return

        user_data = (callback, args)

        self._message.connect(signal_name, self._on_connect_callback, user_data)

    def _on_connect_callback(self, _message: Soup.Message, *args: Any) -> None:
        signal_args = args[:-1]
        callback, user_data = args[-1]
        callback(self, *signal_args, *user_data)

    def send(
        self,
        method: HTTP_METHODS_T,
        uri_string: str,
        timeout: int | None = None,
        callback: Callable[[HTTPRequest], Any] | None = None,
    ) -> None:

        if callback:
            self.connect("finished", callback)
        self._send(method, uri_string, timeout)

    def _send(
        self, method: HTTP_METHODS_T, uri_string: str, timeout: int | None = None
    ) -> None:

        if self._is_finished:
            raise ValueError("Session already finished")

        self._message.set_method(method)

        if self._response_body_file is not None:
            self._output_stream = self._response_body_file.replace(
                None, False, Gio.FileCreateFlags.REPLACE_DESTINATION, self._cancellable
            )

        uri = GLib.Uri.parse(uri_string, GLib.UriFlags(Soup.HTTP_URI_FLAGS))
        self._message.set_uri(uri)

        if self._request_body_data is not None:
            self._message.set_request_body_from_bytes(
                self._request_content_type, GLib.Bytes.new(self._request_body_data)
            )
            if self._emit_request_progress:
                self._message.connect("wrote-body-data", self._on_request_body_progress)

        if self._request_body_file is not None:
            request_input_stream = self._request_body_file.read(self._cancellable)
            self._message.set_request_body(
                self._request_content_type,
                request_input_stream,
                self._request_content_length,
            )
            if self._emit_request_progress:
                self._message.connect("wrote-body-data", self._on_request_body_progress)

        self._message.connect("content-sniffed", self._on_content_sniffed)
        self._message.connect("got-body", self._on_got_body)
        self._message.connect("restarted", self._on_restarted)
        self._message.connect("finished", self._on_finished)
        self._message.connect("got-headers", self._on_got_headers)
        if self._accept_certificate_func is not None:
            self._message.connect("accept-certificate", self._accept_certificate_func)

        soup_session = self._session.get_soup_session()
        soup_session.send_async(
            self._message, GLib.PRIORITY_DEFAULT, self._cancellable, self._on_response
        )

        if timeout is not None:
            self._timeout_id = GLib.timeout_add_seconds(timeout, self._on_timeout)

        self._log.info("Request sent, method: %s, uri: %s", method, uri_string)

    def _on_request_body_progress(self, _message: Soup.Message, chunk: int) -> None:

        self._sent_size += chunk
        self.emit("request-progress", self._sent_size / self._request_content_length)

    def _on_timeout(self) -> None:
        self._timeout_reached = True
        self._timeout_id = None
        self._set_error(HTTPRequestError.TIMEOUT)
        self.cancel()

    def _on_response(self, session: Soup.Session, result: Gio.AsyncResult) -> None:

        self._log.info("HTTP version: %s", self._message.get_http_version().value_name)
        self._log.info("Request response received")
        try:
            self._input_stream = session.send_finish(result)
        except GLib.Error as error:
            quark = GLib.quark_try_string("g-io-error-quark")
            if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
                self._set_failed(HTTPRequestError.CANCELLED)
                return

            self._log.error(error)
            self._set_failed(HTTPRequestError.UNKNOWN)
            return

        status = self._message.get_status()
        if status not in (Soup.Status.OK, Soup.Status.CREATED):
            self._log.info(
                "Response status: %s %s", int(status), self._message.get_reason_phrase()
            )
            self._set_failed(HTTPRequestError.STATUS_NOT_OK)
            return

        self._log.info("Start downloading response body")
        self.emit("starting-response-body")

        self._read_async()

    def _read_async(self) -> None:
        assert self._input_stream is not None
        self._input_stream.read_bytes_async(
            CHUNK_SIZE, GLib.PRIORITY_LOW, self._cancellable, self._on_bytes_read_result
        )

    def _on_bytes_read_result(
        self, input_stream: Gio.InputStream, result: Gio.AsyncResult
    ) -> None:

        try:
            data = input_stream.read_bytes_finish(result)
        except GLib.Error as error:
            self._log.error(error)
            quark = GLib.quark_try_string("g-io-error-quark")
            if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
                self._finish_read(HTTPRequestError.CANCELLED)
            else:
                self._finish_read(HTTPRequestError.UNKNOWN)

            return

        bytes_ = data.get_data()
        if not bytes_:
            self._finish_read()
            return

        self._received_size += len(bytes_)
        if self._check_content_overflow():
            return

        if self._output_stream is None:
            self._response_body_data += bytes_

        else:
            try:
                self._output_stream.write_all(bytes_, self._cancellable)
            except GLib.Error as error:
                self._log.error(error)
                quark = GLib.quark_try_string("g-io-error-quark")
                if error.matches(quark, Gio.IOErrorEnum.CANCELLED):
                    self._finish_read(HTTPRequestError.CANCELLED)
                else:
                    self._finish_read(HTTPRequestError.UNKNOWN)
                return

        self._read_async()
        self._emit_progress()

    def _finish_read(self, error: HTTPRequestError | None = None) -> None:
        self._log.info("Finished reading")
        if error is None:
            self._close_all_streams()
            return

        self._set_failed(error)

    def _on_content_sniffed(
        self,
        message: Soup.Message,
        content_type: str | None,
        _params: GLib.HashTable,
    ) -> None:

        # Signal is only raised when there is content in the response
        headers = message.get_response_headers()
        if content_type is None:
            # According to the docs, content_type is None when the sniffer
            # decides to trust the content-type sent by the server.
            content_type, _ = headers.get_content_type()

        self._response_content_type = content_type or ""

        self._log.info("Sniffed: content-type: %s", self._response_content_type)

        self.emit(
            "content-sniffed",
            self._response_content_length,
            self._response_content_type,
        )

    def _on_got_headers(self, message: Soup.Message) -> None:
        headers = message.get_response_headers()
        self._response_content_length = headers.get_content_length()
        self._log.info("Got Headers: content-length: %s", self._response_content_length)

    def _on_got_body(self, _message: Soup.Message) -> None:
        # This signal tells us that the full body was received.
        # The `finished` signal is not a sure indicator if the message body
        # was received in full, as its also triggered when a message is
        # cancelled.
        self._log.info("Body received")
        self._body_received = True

    def _emit_progress(self) -> None:
        if not self._emit_response_progress:
            return

        if not self._message.get_method() == "GET":
            return

        self.emit(
            "response-progress", self._received_size / self._response_content_length
        )

    def _check_content_overflow(self) -> bool:
        if self._received_size > self._response_content_length:
            self._finish_read(HTTPRequestError.CONTENT_OVERFLOW)
            return True
        return False

    def _on_restarted(self, _message: Soup.Message) -> None:
        self._log.info("Restarted")
        self._body_received = False
        self._response_content_type = ""
        self._response_content_length = 0
        self._received_size = 0

    def _on_finished(self, _message: Soup.Message) -> None:
        self._log.info("Message finished")
        if not self._body_received:
            # This can happen when the message is cancelled. The `finished`
            # signal is raised whenever the input stream is closed.
            # In the case the message was cancelled other parts of the code
            # will call set_failed().
            return

        if self._received_size != self._response_content_length:
            self._set_failed(HTTPRequestError.INCOMPLETE)
            return

        status = self._message.get_status()
        if status not in (Soup.Status.OK, Soup.Status.CREATED):
            self._set_failed(HTTPRequestError.STATUS_NOT_OK)
            return

        self._log.info("Request status: %s", Soup.Status.get_phrase(status))

        if self._cancellable.is_cancelled():
            # It can happen that the message is finished before the
            # response callback returns after calling cancel(). If
            # we call complete, the response callback will also
            # try to cleanup and will fail.
            self._log.info(
                "Skip setting message complete because " "cancel is in progess"
            )
            return

        self._set_complete()

    def _set_error(self, error: HTTPRequestError) -> None:
        self._log.info("Set Error: %s", error)
        self._error = error

    def _set_failed(self, error: HTTPRequestError) -> None:
        self._log.info("Set Failed: %s", error)
        self._is_finished = True

        if self._error is None:
            self._set_error(error)

        self._close_all_streams()
        self.emit("finished")
        self._cleanup()

    def _set_complete(self) -> None:
        self._log.info("Set Complete")
        self._is_finished = True
        self._is_complete = True
        self._close_all_streams()
        self.emit("finished")
        self._cleanup()

    def _close_all_streams(self) -> None:
        # stream.close() will invoke signals on the Message object
        # which in turn can lead to this method called again in the
        # same Mainloop iteration. This means is_closed() will not
        # return True and we get an GLib.IOError.PENDING error.

        input_stream = self._input_stream
        output_stream = self._output_stream

        self._input_stream = None
        self._output_stream = None

        if input_stream is not None:
            if not input_stream.is_closed():
                input_stream.close(None)

        if output_stream is not None:
            if not output_stream.is_closed():
                output_stream.close(None)

    def _cleanup(self) -> None:
        self._log.info("Run cleanup")

        self._response_body_data = b""
        self._message.run_dispose()

        del self._cancellable
        del self._user_data

        if self._timeout_id is not None:
            GLib.source_remove(self._timeout_id)
            self._timeout_id = None

        self.emit("destroy")
        self.run_dispose()

    def __repr__(self) -> str:
        return f"Request({id(self)})"