text: # -*- coding: utf-8 -*-
  # cython: language_level=3
  # Copyright (c) 2020 Nekokatt
  # Copyright (c) 2021-present davfsa
  #
  # Permission is hereby granted, free of charge, to any person obtaining a copy
  # of this software and associated documentation files (the "Software"), to deal
  # in the Software without restriction, including without limitation the rights
  # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  # copies of the Software, and to permit persons to whom the Software is
  # furnished to do so, subject to the following conditions:
  #
  # The above copyright notice and this permission notice shall be included in all
  # copies or substantial portions of the Software.
  #
  # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  # SOFTWARE.
  """Standard implementation of a REST based interactions server."""
  from __future__ import annotations
  
  __all__: typing.Sequence[str] = ("InteractionServer",)
  
  import asyncio
  import logging
  import threading
  import typing
  
  import aiohttp
  import aiohttp.web
  import aiohttp.web_runner
  
  from hikari import applications
  from hikari import errors
  from hikari.api import interaction_server
  from hikari.api import special_endpoints
  from hikari.interactions import base_interactions
  from hikari.internal import data_binding
  
  if typing.TYPE_CHECKING:
      import concurrent.futures
      import socket as socket_
      import ssl
  
      import aiohttp.abc
      import aiohttp.typedefs
  
      # This is kept inline as pynacl is an optional dependency.
      from nacl import signing
  
      from hikari import files as files_
      from hikari.api import entity_factory as entity_factory_api
      from hikari.api import rest as rest_api
      from hikari.interactions import command_interactions
      from hikari.interactions import component_interactions
  
      _InteractionT_co = typing.TypeVar("_InteractionT_co", bound=base_interactions.PartialInteraction, covariant=True)
      _MessageResponseBuilderT = typing.Union[
          special_endpoints.InteractionDeferredBuilder,
          special_endpoints.InteractionMessageBuilder,
      ]
  
  _LOGGER: typing.Final[logging.Logger] = logging.getLogger("hikari.interaction_server")
  
  # Internal interaction and interaction response types.
  _PING_INTERACTION_TYPE: typing.Final[int] = 1
  _PONG_RESPONSE_TYPE: typing.Final[int] = 1
  
  # HTTP status codes.
  _OK_STATUS: typing.Final[int] = 200
  _BAD_REQUEST_STATUS: typing.Final[int] = 400
  _PAYLOAD_TOO_LARGE_STATUS: typing.Final[int] = 413
  _UNSUPPORTED_MEDIA_TYPE_STATUS: typing.Final[int] = 415
  _INTERNAL_SERVER_ERROR_STATUS: typing.Final[int] = 500
  _NOT_IMPLEMENTED: typing.Final[int] = 501
  
  _UTF_8_CHARSET: typing.Final[str] = "UTF-8"
  
  # Header keys and values
  _X_SIGNATURE_ED25519_HEADER: typing.Final[str] = "X-Signature-Ed25519"
  _X_SIGNATURE_TIMESTAMP_HEADER: typing.Final[str] = "X-Signature-Timestamp"
  _CONTENT_TYPE_KEY: typing.Final[str] = "Content-Type"
  _USER_AGENT_KEY: typing.Final[str] = "User-Agent"
  _APPLICATION_OCTET_STREAM: typing.Final[str] = "application/octet-stream"
  _JSON_CONTENT_TYPE: typing.Final[str] = "application/json"
  _JSON_TYPE_WITH_CHARSET: typing.Final[str] = f"{_JSON_CONTENT_TYPE}; charset={_UTF_8_CHARSET}"
  _TEXT_CONTENT_TYPE: typing.Final[str] = "text/plain"
  _TEXT_TYPE_WITH_CHARSET: typing.Final[str] = f"{_TEXT_CONTENT_TYPE}; charset={_UTF_8_CHARSET}"
  
  
  class _Response:
      __slots__: typing.Sequence[str] = ("_content_type", "_files", "_payload", "_status_code")
  
      def __init__(
          self,
          status_code: int,
          payload: typing.Optional[bytes] = None,
          *,
          content_type: typing.Optional[str] = None,
          files: typing.Sequence[files_.Resource[files_.AsyncReader]] = (),
      ) -> None:
          if payload and not content_type:
              content_type = _TEXT_TYPE_WITH_CHARSET
  
          self._content_type = content_type
          self._files = files
          self._payload = payload
          self._status_code = status_code
  
      @property
      def content_type(self) -> typing.Optional[str]:
          return self._content_type
  
      @property
      def files(self) -> typing.Sequence[files_.Resource[files_.AsyncReader]]:
          return self._files
  
      @property
      def headers(self) -> typing.Optional[typing.MutableMapping[str, str]]:
          return None
  
      @property
      def payload(self) -> typing.Optional[bytes]:
          return self._payload
  
      @property
      def status_code(self) -> int:
          return self._status_code
  
  
  # Constant response
  _PONG_RESPONSE: typing.Final[_Response] = _Response(
      _OK_STATUS, data_binding.dump_json({"type": _PONG_RESPONSE_TYPE}).encode(), content_type=_JSON_TYPE_WITH_CHARSET
  )
  
  
  class _FilePayload(aiohttp.Payload):
      _value: files_.Resource[files_.AsyncReader]
  
      def __init__(
          self,
          value: files_.Resource[files_.AsyncReader],
          content_type: str,
          /,
          *,
          executor: typing.Optional[concurrent.futures.Executor] = None,
          headers: typing.Optional[typing.Dict[str, str]] = None,
      ) -> None:
          super().__init__(value=value, headers=headers, content_type=content_type)
          self._executor = executor
  
      async def write(self, writer: aiohttp.abc.AbstractStreamWriter) -> None:
          async with self._value.stream(executor=self._executor) as data:
              async for chunk in data:
                  await writer.write(chunk)
  
  
  class InteractionServer(interaction_server.InteractionServer):
      """Standard implementation of `hikari.api.interaction_server.InteractionServer`.
  
      Parameters
      ----------
      entity_factory : hikari.api.entity_factory.EntityFactory
          The entity factory instance this server should use.
  
      Other Parameters
      ----------------
      dumps : aiohttp.typedefs.JSONEncoder
          The JSON encoder this server should use. Defaults to `json.dumps`.
      loads : aiohttp.typedefs.JSONDecoder
          The JSON decoder this server should use. Defaults to `json.loads`.
      public_key : builtins.bytes
          The public key this server should use for verifying request payloads from
          Discord. If left as `builtins.None` then the client will try to work this
          out using `rest_client`.
      rest_client : hikari.api.rest.RESTClient
          The client this should use for making REST requests.
      """
  
      __slots__: typing.Sequence[str] = (
          "_application_fetch_lock",
          "_close_event",
          "_dumps",
          "_entity_factory",
          "_executor",
          "_is_closing",
          "_listeners",
          "_loads",
          "_nacl",
          "_public_key",
          "_rest_client",
          "_server",
      )
  
      def __init__(
          self,
          *,
          dumps: aiohttp.typedefs.JSONEncoder = data_binding.dump_json,
          entity_factory: entity_factory_api.EntityFactory,
          executor: typing.Optional[concurrent.futures.Executor] = None,
          loads: aiohttp.typedefs.JSONDecoder = data_binding.load_json,
          rest_client: rest_api.RESTClient,
          public_key: typing.Optional[bytes] = None,
      ) -> None:
          # This is kept inline as pynacl is an optional dependency.
          try:
              import nacl.exceptions
              import nacl.signing
  
          except ModuleNotFoundError as exc:
              raise RuntimeError(
                  "You must install the optional `hikari[server]` dependencies to use the default interaction server."
              ) from exc
  
          # Building asyncio.Lock when there isn't a running loop may lead to runtime errors.
          self._application_fetch_lock: typing.Optional[asyncio.Lock] = None
          # Building asyncio.Event when there isn't a running loop may lead to runtime errors.
          self._close_event: typing.Optional[asyncio.Event] = None
          self._dumps = dumps
          self._entity_factory = entity_factory
          self._executor = executor
          self._is_closing = False
          self._listeners: typing.Dict[typing.Type[base_interactions.PartialInteraction], typing.Any] = {}
          self._loads = loads
          self._nacl = nacl
          self._rest_client = rest_client
          self._server: typing.Optional[aiohttp.web_runner.AppRunner] = None
          self._public_key = nacl.signing.VerifyKey(public_key) if public_key is not None else None
  
      @property
      def is_alive(self) -> bool:
          """Whether this interaction server is active.
  
          Returns
          -------
          builtins.bool
              Whether this interaction server is active
          """
          return self._server is not None
  
      async def _fetch_public_key(self) -> signing.VerifyKey:
          if self._application_fetch_lock is None:
              self._application_fetch_lock = asyncio.Lock()
  
          application: typing.Union[applications.Application, applications.AuthorizationApplication]
          async with self._application_fetch_lock:
              if self._public_key:
                  return self._public_key
  
              if self._rest_client.token_type == applications.TokenType.BOT:
                  application = await self._rest_client.fetch_application()
  
              else:
                  application = (await self._rest_client.fetch_authorization()).application
  
              self._public_key = self._nacl.signing.VerifyKey(application.public_key)
              return self._public_key
  
      async def aiohttp_hook(self, request: aiohttp.web.Request) -> aiohttp.web.Response:
          """Handle an AIOHTTP interaction request.
  
          This method handles aiohttp specific detail before calling
          `InteractionServer.on_interaction` with the data extracted from the
          request if it can and handles building an aiohttp response.
  
          Parameters
          ----------
          request : aiohttp.web.Request
              The received request.
  
          Returns
          -------
          aiohttp.web.Response
              The aiohttp response.
          """
          if request.content_type.lower() != _JSON_CONTENT_TYPE:
              _LOGGER.debug("Payload with invalid media type %r received", request.content_type)
              return aiohttp.web.Response(
                  status=_UNSUPPORTED_MEDIA_TYPE_STATUS,
                  body=b"Unsupported Media Type",
                  content_type=_TEXT_CONTENT_TYPE,
                  charset=_UTF_8_CHARSET,
              )
  
          try:
              signature_header = bytes.fromhex(request.headers[_X_SIGNATURE_ED25519_HEADER])
              timestamp_header = request.headers[_X_SIGNATURE_TIMESTAMP_HEADER].encode()
  
          except (KeyError, ValueError):
              user_agent = request.headers.get(_USER_AGENT_KEY, "NONE")
              _LOGGER.debug("Received a request with a missing or invalid signature header (UA %r)", user_agent)
              return aiohttp.web.Response(
                  status=_BAD_REQUEST_STATUS,
                  body=b"Missing or invalid required request signature header(s)",
                  content_type=_TEXT_CONTENT_TYPE,
                  charset=_UTF_8_CHARSET,
              )
  
          try:
              body = await request.read()
  
          except aiohttp.web.HTTPRequestEntityTooLarge:
              _LOGGER.debug("Received a request with a payload that's too large to process")
              return aiohttp.web.Response(
                  status=_PAYLOAD_TOO_LARGE_STATUS,
                  body=b"Payload too large",
                  content_type=_TEXT_CONTENT_TYPE,
                  charset=_UTF_8_CHARSET,
              )
  
          if not body:
              user_agent = request.headers.get(_USER_AGENT_KEY, "NONE")
              _LOGGER.debug("Received a body-less request (UA %r)", user_agent)
              return aiohttp.web.Response(
                  status=_BAD_REQUEST_STATUS,
                  body=b"POST request must have a body",
                  content_type=_TEXT_CONTENT_TYPE,
                  charset=_UTF_8_CHARSET,
              )
  
          response = await self.on_interaction(body=body, signature=signature_header, timestamp=timestamp_header)
  
          if response.files:
              multipart = aiohttp.MultipartWriter(subtype="form-data")
              if response.payload:
                  body_payload = aiohttp.BytesPayload(response.payload, content_type=response.content_type)
                  body_payload.set_content_disposition("form-data", name="payload_json")
                  multipart.append_payload(body_payload)
  
              for index, file_ in enumerate(response.files):
                  async with file_.stream(head_only=True) as stream:
                      mimetype = stream.mimetype or _APPLICATION_OCTET_STREAM
  
                  payload = _FilePayload(file_, mimetype, executor=self._executor)
                  payload.set_content_disposition("form-data", name=f"files[{index}]", filename=file_.filename)
                  multipart.append_payload(payload)
  
              return aiohttp.web.Response(status=response.status_code, headers=response.headers, body=multipart)
  
          headers = response.headers
          if response.content_type:
              headers = headers or {}
              headers[_CONTENT_TYPE_KEY] = response.content_type
  
          return aiohttp.web.Response(status=response.status_code, headers=headers, body=response.payload)
  
      async def close(self) -> None:
          """Gracefully close the server and any open connections."""
          if not self._server or not self._close_event:
              raise errors.ComponentStateConflictError("Cannot close an inactive interaction server")
  
          if self._is_closing:
              await self.join()
              return
  
          self._is_closing = True
          self._application_fetch_lock = None
          # This shutdown then cleanup ordering matters.
          await self._server.shutdown()
          await self._server.cleanup()
          self._close_event.set()
          self._close_event = None
          self._server = None
  
      async def join(self) -> None:
          """Wait for the process to halt before continuing."""
          if not self._close_event:
              raise errors.ComponentStateConflictError("Cannot wait for an inactive interaction server to join")
  
          await self._close_event.wait()
  
      async def on_interaction(self, body: bytes, signature: bytes, timestamp: bytes) -> interaction_server.Response:
          """Handle an interaction received from Discord as a REST server.
  
          !!! note
              If this server instance is alive then this will be called internally
              by the server but if the instance isn't alive then this may still be
              called externally to trigger interaction dispatch.
  
          Parameters
          ----------
          body : builtins.bytes
              The interaction payload.
          signature : builtins.bytes
              Value of the `"X-Signature-Ed25519"` header used to verify the body.
          timestamp : builtins.bytes
              Value of the `"X-Signature-Timestamp"` header used to verify the body.
  
          Returns
          -------
          hikari.api.interaction_server.Response
              Instructions on how the REST server calling this should respond to
              the interaction request.
          """
          public_key = self._public_key or await self._fetch_public_key()
  
          try:
              public_key.verify(timestamp + body, signature)
  
          except (self._nacl.exceptions.BadSignatureError, ValueError):
              _LOGGER.error("Received a request with an invalid signature")
              return _Response(_BAD_REQUEST_STATUS, b"Invalid request signature")
  
          try:
              payload = self._loads(body.decode("utf-8"))
              interaction_type = int(payload["type"])
  
          except (data_binding.JSONDecodeError, ValueError, TypeError) as exc:
              _LOGGER.error("Received a request with an invalid JSON body", exc_info=exc)
              return _Response(_BAD_REQUEST_STATUS, b"Invalid JSON body")
  
          except KeyError as exc:
              _LOGGER.error("Missing 'type' field in received JSON payload", exc_info=exc)
              return _Response(_BAD_REQUEST_STATUS, b"Missing required 'type' field in payload")
  
          if interaction_type == _PING_INTERACTION_TYPE:
              _LOGGER.debug("Responding to ping interaction")
              return _PONG_RESPONSE
  
          try:
              interaction = self._entity_factory.deserialize_interaction(payload)
  
          except errors.UnrecognisedEntityError:
              _LOGGER.debug("Ignoring unknown interaction type %s", interaction_type)
              return _Response(_NOT_IMPLEMENTED, b"Interaction type not implemented")
  
          except Exception as exc:
              asyncio.get_running_loop().call_exception_handler(
                  {"message": "Exception occurred during interaction deserialization", "exception": exc}
              )
              return _Response(_INTERNAL_SERVER_ERROR_STATUS, b"Exception occurred during interaction deserialization")
  
          if listener := self._listeners.get(type(interaction)):
              _LOGGER.debug("Dispatching interaction %s", interaction.id)
              try:
                  result = await listener(interaction)
                  raw_payload, files = result.build(self._entity_factory)
                  payload = self._dumps(raw_payload)
  
              except Exception as exc:
                  asyncio.get_running_loop().call_exception_handler(
                      {"message": "Exception occurred during interaction dispatch", "exception": exc}
                  )
                  return _Response(_INTERNAL_SERVER_ERROR_STATUS, b"Exception occurred during interaction dispatch")
  
              return _Response(_OK_STATUS, payload.encode(), files=files, content_type=_JSON_TYPE_WITH_CHARSET)
  
          _LOGGER.debug(
              "Ignoring interaction %s of type %s without registered listener", interaction.id, interaction.type
          )
          return _Response(_NOT_IMPLEMENTED, b"Handler not set for this interaction type")
  
      async def start(
          self,
          backlog: int = 128,
          enable_signal_handlers: typing.Optional[bool] = None,
          host: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
          port: typing.Optional[int] = None,
          path: typing.Optional[str] = None,
          reuse_address: typing.Optional[bool] = None,
          reuse_port: typing.Optional[bool] = None,
          socket: typing.Optional[socket_.socket] = None,
          shutdown_timeout: float = 60.0,
          ssl_context: typing.Optional[ssl.SSLContext] = None,
      ) -> None:
          """Start the bot and wait for the internal server to startup then return.
  
          Other Parameters
          ----------------
          backlog : builtins.int
              The number of unaccepted connections that the system will allow before
              refusing new connections.
          enable_signal_handlers : typing.Optional[builtins.bool]
              Defaults to `builtins.True` if this is started in the main thread.
  
              If on a __non-Windows__ OS with builtin support for kernel-level
              POSIX signals, then setting this to `builtins.True` will allow
              treating keyboard interrupts and other OS signals to safely shut
              down the application as calls to shut down the application properly
              rather than just killing the process in a dirty state immediately.
              You should leave this enabled unless you plan to implement your own
              signal handling yourself.
          host : typing.Optional[typing.Union[builtins.str, aiohttp.web.HostSequence]]
              TCP/IP host or a sequence of hosts for the HTTP server.
          port : typing.Optional[builtins.int]
              TCP/IP port for the HTTP server.
          path : typing.Optional[builtins.str]
              File system path for HTTP server unix domain socket.
          reuse_address : typing.Optional[builtins.bool]
              Tells the kernel to reuse a local socket in TIME_WAIT state, without
              waiting for its natural timeout to expire.
          reuse_port : typing.Optional[builtins.bool]
              Tells the kernel to allow this endpoint to be bound to the same port
              as other existing endpoints are also bound to.
          socket : typing.Optional[socket.socket]
              A pre-existing socket object to accept connections on.
          shutdown_timeout : builtins.float
              A delay to wait for graceful server shutdown before forcefully
              disconnecting all open client sockets. This defaults to 60 seconds.
          ssl_context : typing.Optional[ssl.SSLContext]
              SSL context for HTTPS servers.
  
          !!! note
              For more information on the other parameters such as defaults see
              AIOHTTP's documentation.
          """
          if self._server:
              raise errors.ComponentStateConflictError("Cannot start an already active interaction server")
  
          if enable_signal_handlers is None:
              enable_signal_handlers = threading.current_thread() is threading.main_thread()
  
          self._close_event = asyncio.Event()
          self._is_closing = False
          aio_app = aiohttp.web.Application()
          aio_app.add_routes([aiohttp.web.post("/", self.aiohttp_hook)])
          self._server = aiohttp.web_runner.AppRunner(aio_app, handle_signals=enable_signal_handlers, access_log=_LOGGER)
          await self._server.setup()
          sites: typing.List[aiohttp.web.BaseSite] = []
  
          if host is not None:
              if isinstance(host, str):
                  host = [host]
  
              for h in host:
                  sites.append(
                      aiohttp.web.TCPSite(
                          self._server,
                          h,
                          port,
                          shutdown_timeout=shutdown_timeout,
                          ssl_context=ssl_context,
                          backlog=backlog,
                          reuse_address=reuse_address,
                          reuse_port=reuse_port,
                      )
                  )
  
          elif path is None and socket is None or port is None:
              sites.append(
                  aiohttp.web.TCPSite(
                      self._server,
                      port=port,
                      shutdown_timeout=shutdown_timeout,
                      ssl_context=ssl_context,
                      backlog=backlog,
                      reuse_address=reuse_address,
                      reuse_port=reuse_port,
                  )
              )
  
          if path is not None:
              sites.append(
                  aiohttp.web.UnixSite(
                      self._server, path, shutdown_timeout=shutdown_timeout, ssl_context=ssl_context, backlog=backlog
                  )
              )
  
          if socket is not None:
              sites.append(
                  aiohttp.web.SockSite(
                      self._server, socket, shutdown_timeout=shutdown_timeout, ssl_context=ssl_context, backlog=backlog
                  )
              )
  
          for site in sites:
              _LOGGER.info("Starting site on %s", site.name)
              await site.start()
  
      @typing.overload
      def get_listener(
          self, interaction_type: typing.Type[command_interactions.CommandInteraction], /
      ) -> typing.Optional[
          interaction_server.ListenerT[command_interactions.CommandInteraction, _MessageResponseBuilderT]
      ]:
          ...
  
      @typing.overload
      def get_listener(
          self, interaction_type: typing.Type[component_interactions.ComponentInteraction], /
      ) -> typing.Optional[
          interaction_server.ListenerT[component_interactions.ComponentInteraction, _MessageResponseBuilderT]
      ]:
          ...
  
      @typing.overload
      def get_listener(
          self, interaction_type: typing.Type[command_interactions.AutocompleteInteraction], /
      ) -> typing.Optional[
          interaction_server.ListenerT[
              command_interactions.AutocompleteInteraction, special_endpoints.InteractionAutocompleteBuilder
          ]
      ]:
          ...
  
      @typing.overload
      def get_listener(
          self, interaction_type: typing.Type[_InteractionT_co], /
      ) -> typing.Optional[interaction_server.ListenerT[_InteractionT_co, special_endpoints.InteractionResponseBuilder]]:
          ...
  
      def get_listener(
          self, interaction_type: typing.Type[_InteractionT_co], /
      ) -> typing.Optional[interaction_server.ListenerT[_InteractionT_co, special_endpoints.InteractionResponseBuilder]]:
          return self._listeners.get(interaction_type)
  
      @typing.overload
      def set_listener(
          self,
          interaction_type: typing.Type[command_interactions.CommandInteraction],
          listener: typing.Optional[
              interaction_server.ListenerT[command_interactions.CommandInteraction, _MessageResponseBuilderT]
          ],
          /,
          *,
          replace: bool = False,
      ) -> None:
          ...
  
      @typing.overload
      def set_listener(
          self,
          interaction_type: typing.Type[component_interactions.ComponentInteraction],
          listener: typing.Optional[
              interaction_server.ListenerT[component_interactions.ComponentInteraction, _MessageResponseBuilderT]
          ],
          /,
          *,
          replace: bool = False,
      ) -> None:
          ...
  
      @typing.overload
      def set_listener(
          self,
          interaction_type: typing.Type[command_interactions.AutocompleteInteraction],
          listener: typing.Optional[
              interaction_server.ListenerT[
                  command_interactions.AutocompleteInteraction, special_endpoints.InteractionAutocompleteBuilder
              ]
          ],
          /,
          *,
          replace: bool = False,
      ) -> None:
          ...
  
      def set_listener(
          self,
          interaction_type: typing.Type[_InteractionT_co],
          listener: typing.Optional[
              interaction_server.ListenerT[_InteractionT_co, special_endpoints.InteractionResponseBuilder]
          ],
          /,
          *,
          replace: bool = False,
      ) -> None:
          if listener:
              if not replace and interaction_type in self._listeners:
                  raise TypeError(f"Listener already set for {interaction_type.__name__}")
  
              self._listeners[interaction_type] = listener
  
          else:
              self._listeners.pop(interaction_type, None)
  