# Copyright(c) 2015-2024 ACCESS CO., LTD. All rights reserved.

use Croma

defmodule AntikytheraCore.Handler.GearAction.Web do
  alias Croma.Result, as: R
  alias Antikythera.{GearName, PathInfo, Conn, GearActionTimeout}
  alias Antikythera.Http.{Method, QueryParams, Body}
  alias Antikythera.Request.PathMatches
  alias Antikythera.Context.GearEntryPoint
  alias AntikytheraCore.Handler.GearAction
  alias AntikytheraCore.Handler.GearError
  alias AntikytheraCore.Handler.HelperModules
  alias AntikytheraCore.Handler.CowboyReq
  alias AntikytheraCore.Handler.ExecutorPoolHelper
  alias AntikytheraCore.Handler.WebsocketState
  alias AntikytheraCore.Conn, as: CoreConn
  alias AntikytheraCore.GearModule
  alias AntikytheraCore.ExecutorPool.ActionRunner
  alias AntikytheraCore.GearLog.ContextHelper

  # This line results in "conflicting behaviours - callback init/2 required by both 'cowboy_websocket' and 'cowboy_handler'"
  # @behaviour :cowboy_handler
  @behaviour :cowboy_websocket

  @type http_reply :: {:ok, :cowboy_req.req(), nil}
  @type ws_upgrade :: {:cowboy_websocket, :cowboy_req.req(), Conn.t(), timeout}

  # (Should we make this a mix config item?)
  max_frame_size = 5_000_000
  @ws_upgrade_options %{idle_timeout: 60_000, compress: true, max_frame_size: max_frame_size}

  @impl true
  defun init(req1 :: :cowboy_req.req(), gear_name :: v[GearName.t()]) :: http_reply | ws_upgrade do
    R.m do
      method <- CowboyReq.method(req1)
      path_info = CowboyReq.path_info(req1)
      helper_modules = GearModule.request_helper_modules(gear_name)

      {entry_point, path_matches, ws?, timeout} <-
        find_route(req1, gear_name, method, path_info, helper_modules)

      routing_info = {gear_name, entry_point, method, path_info, path_matches}
      qparams <- CowboyReq.query_params(req1, routing_info)

      {req2, body_pair} <-
        CowboyReq.request_body_pair(req1, routing_info, qparams, helper_modules)

      pure(
        run_action_with_conn(req2, routing_info, qparams, body_pair, helper_modules, ws?, timeout)
      )
    end
    |> case do
      # protocol upgrade to websocket
      {:ok, {req3, state}} -> {:cowboy_websocket, req3, state, @ws_upgrade_options}
      # normal response
      {:ok, req_reply} -> {:ok, req_reply, nil}
      # error response
      {:error, req_reply} -> {:ok, req_reply, nil}
    end
  end

  defunp find_route(
           req :: :cowboy_req.req(),
           gear_name :: v[GearName.t()],
           method :: v[Method.t()],
           path_info :: v[PathInfo.t()],
           %HelperModules{router: router}
         ) :: R.t({GearEntryPoint.t(), PathMatches.t(), boolean}) do
    case router.__web_route__(method, path_info) do
      {controller, action, path_matches, websocket?, timeout} ->
        {:ok, {{controller, action}, path_matches, websocket?, timeout}}

      nil ->
        {:error,
         CowboyReq.with_conn(
           req,
           {gear_name, nil, method, path_info, %{}},
           %{},
           fn conn, _context -> GearError.no_route(conn) end
         )}
    end
  end

  defunp run_action_with_conn(
           req :: :cowboy_req.req(),
           routing_info :: CowboyReq.routing_info(),
           qparams :: v[QueryParams.t()],
           body_pair :: {binary, Body.t()},
           helper_modules :: v[HelperModules.t()],
           websocket? :: v[boolean],
           timeout :: v[GearActionTimeout.t()]
         ) :: :cowboy_req.req() | {:cowboy_req.req(), WebsocketState.t()} do
    case websocket? do
      true ->
        run_action_with_conn_ws(req, routing_info, qparams, body_pair, helper_modules, timeout)

      false ->
        run_action_with_conn_http(req, routing_info, qparams, body_pair, helper_modules, timeout)
    end
  end

  defunp run_action_with_conn_http(
           req :: :cowboy_req.req(),
           {gear_name, entry_point, _, _, _} = routing_info :: CowboyReq.routing_info(),
           qparams :: v[QueryParams.t()],
           body_pair :: {binary, Body.t()},
           helper_modules :: v[HelperModules.t()],
           timeout :: v[GearActionTimeout.t()]
         ) :: :cowboy_req.req() do
    CowboyReq.with_conn(req, routing_info, qparams, body_pair, fn conn, context ->
      GearAction.with_logging_and_metrics_reporting(conn, context, helper_modules, fn ->
        run_action_with_executor(conn, gear_name, entry_point, helper_modules, timeout)
      end)
    end)
  end

  defunp run_action_with_conn_ws(
           req :: :cowboy_req.req(),
           {gear_name, entry_point, _, _, _} = routing_info :: CowboyReq.routing_info(),
           qparams :: v[QueryParams.t()],
           body_pair :: {binary, Body.t()},
           helper_modules :: v[HelperModules.t()],
           timeout :: v[GearActionTimeout.t()]
         ) :: :cowboy_req.req() | {:cowboy_req.req(), WebsocketState.t()} do
    context = GearAction.Context.make()
    conn1 = CoreConn.make_from_cowboy_req(req, routing_info, qparams, body_pair)
    ContextHelper.set(conn1)

    GearAction.with_logging_and_metrics_reporting(conn1, context, helper_modules, fn ->
      case run_action_with_executor(conn1, gear_name, entry_point, helper_modules, timeout) do
        # Fill the status code with "101 Upgrade" in order to correctly report response metrics
        conn2 = %Conn{status: nil} -> %Conn{conn2 | status: 101}
        conn2 -> conn2
      end
    end)
    |> case do
      %Conn{status: 101} = conn3 ->
        ExecutorPoolHelper.increment_ws_count(
          conn3,
          context.start_time_for_log,
          req,
          helper_modules,
          fn ->
            {req,
             WebsocketState.make(conn3, context.start_time_for_log, entry_point, helper_modules)}
          end
        )

      conn3 ->
        CoreConn.reply_as_cowboy_res(conn3, req)
    end
  end

  defunp run_action_with_executor(
           conn1 :: v[Conn.t()],
           gear_name :: v[GearName.t()],
           entry_point :: v[GearEntryPoint.t()],
           helper_modules :: v[HelperModules.t()],
           timeout :: v[GearActionTimeout.t()]
         ) :: Conn.t() do
    ExecutorPoolHelper.with_executor(conn1, gear_name, helper_modules, fn pid, conn2 ->
      ActionRunner.run(pid, conn2, entry_point, timeout)
    end)
  end

  #
  # callback implementations for cowboy_websocket
  #
  @impl true
  defun websocket_init(ws_state :: v[WebsocketState.t()]) :: WebsocketState.callback_result() do
    WebsocketState.init(ws_state)
  end

  @impl true
  defun websocket_handle(frame :: :cow_ws.frame(), ws_state :: v[WebsocketState.t()]) ::
          WebsocketState.callback_result() do
    case frame do
      :ping -> {:ok, ws_state}
      :pong -> {:ok, ws_state}
      {ping_or_pong, _payload} when ping_or_pong in [:ping, :pong] -> {:ok, ws_state}
      _ -> WebsocketState.handle_client_message(ws_state, frame)
    end
  end

  @impl true
  defun websocket_info(message :: any, ws_state :: v[WebsocketState.t()]) ::
          WebsocketState.callback_result() do
    case message do
      {:EXIT, _, _} ->
        # In rare conditions, websocket connection process receives an EXIT message
        # about death of the original handler (i.e. the process that executed the `init/2` callback).
        # Just neglect the message.
        {:ok, ws_state}

      {:antikythera_internal, :close} ->
        close_frame = {:close, 1001, "server shutting down; please reconnect"}
        {:reply, [close_frame], ws_state}

      _ ->
        WebsocketState.handle_server_message(ws_state, message)
    end
  end

  @impl true
  def terminate(reason, _maybe_req, %WebsocketState{} = ws_state) do
    WebsocketState.terminate(ws_state, reason)
  end

  def terminate(_reason, _maybe_req, _state) do
    # normal HTTP request, do nothing
    :ok
  end
end
