ex_aliyun_mns

defmodule ExAliyun.MNS.Client do
  @moduledoc false

  require Logger

  alias ExAliyun.MNS.{Xml, Parser, Application}

  def request(%{action: action} = operation, config, http_opts \\ []) do
    operation
    |> send_request(config, http_opts)
    |> parse_response(action)
  end

  defp send_request(
         %{action: "CreateQueue", params: %{queue_name: queue_name} = params} = _operation,
         config,
         http_opts
       ) do
    body = Xml.generate_queue(params)

    config
    |> new_client(http_opts)
    |> put("/queues/#{queue_name}", body)
  end

  defp send_request(
         %{action: "SetQueueAttributes", params: %{queue_url: queue_url} = params},
         config,
         http_opts
       ) do
    body = Xml.generate_queue(params)

    config
    |> new_client(http_opts)
    |> put("#{queue_url}", body, query: [{"metaoverride", true}])
  end

  defp send_request(
         %{action: "GetQueueAttributes", params: %{queue_url: queue_url}},
         config,
         http_opts
       ) do
    config
    |> new_client(http_opts)
    |> get(queue_url)
  end

  defp send_request(%{action: "ListQueues", headers: headers}, config, http_opts) do
    config
    |> new_client(http_opts)
    |> get("/queues", headers: headers)
  end

  defp send_request(%{action: "DeleteQueue", params: %{queue_url: queue_url}}, config, http_opts) do
    config
    |> new_client(http_opts)
    |> delete(queue_url)
  end

  defp send_request(
         %{action: "SendMessage", params: %{queue_url: queue_url} = params},
         config,
         http_opts
       ) do
    body = Xml.generate_message(params)

    config
    |> new_client(http_opts)
    |> post("#{queue_url}/messages", body)
  end

  defp send_request(
         %{action: "BatchSendMessage", params: %{queue_url: queue_url, messages: messages}},
         config,
         http_opts
       ) do
    body = Xml.generate_messages(messages)

    config
    |> new_client(http_opts)
    |> post("#{queue_url}/messages", body)
  end

  defp send_request(
         %{
           action: "DeleteMessage",
           params: %{queue_url: queue_url, receipt_handle: receipt_handle}
         },
         config,
         http_opts
       ) do
    config
    |> new_client(http_opts)
    |> delete("#{queue_url}/messages", query: [{"ReceiptHandle", receipt_handle}])
  end

  defp send_request(
         %{
           action: "BatchDeleteMessage",
           params: %{queue_url: queue_url, receipt_handles: receipt_handles}
         },
         config,
         http_opts
       ) do
    body = Xml.generate_receipt_handles(receipt_handles)

    config
    |> new_client(http_opts)
    |> delete("#{queue_url}/messages", body: body)
  end

  defp send_request(
         %{action: "ReceiveMessage", params: %{queue_url: queue_url} = params},
         config,
         http_opts
       ) do
    number = Map.get(params, :number)
    wait_time_seconds = Map.get(params, :wait_time_seconds)

    query = format_query([{"numOfMessages", number}, {"waitseconds", wait_time_seconds}])

    opts =
      if wait_time_seconds != nil, do: [timeout: (wait_time_seconds + 2) * 1000], else: http_opts

    config
    |> new_client(opts)
    |> get("#{queue_url}/messages", query: query)
  end

  defp send_request(
         %{action: "PeekMessage", params: %{queue_url: queue_url} = params},
         config,
         http_opts
       ) do
    number = Map.get(params, :number)
    query = format_query([{"peekonly", true}, {"numOfMessages", number}])

    config
    |> new_client(http_opts)
    |> get("#{queue_url}/messages", query: query)
  end

  defp send_request(
         %{
           action: "ChangeMessageVisibility",
           params: %{
             queue_url: queue_url,
             receipt_handle: receipt_handle,
             visibility_timeout: visibility_timeout
           }
         },
         config,
         http_opts
       ) do
    query = [{"receiptHandle", receipt_handle}, {"visibilityTimeout", visibility_timeout}]

    config
    |> new_client(http_opts)
    |> put("#{queue_url}/messages", nil, query: query)
  end

  defp send_request(
         %{action: "CreateTopic", params: %{topic_name: topic_name} = params},
         config,
         http_opts
       ) do
    body = Xml.generate_topic(params)

    config
    |> new_client(http_opts)
    |> put("/topics/#{topic_name}", body)
  end

  defp send_request(
         %{action: "SetTopicAttributes", params: %{topic_url: topic_url} = params},
         config,
         http_opts
       ) do
    body = Xml.generate_topic(params)

    config
    |> new_client(http_opts)
    |> put("#{topic_url}", body, query: [{"metaoverride", true}])
  end

  defp send_request(
         %{action: "GetTopicAttributes", params: %{topic_url: topic_url}},
         config,
         http_opts
       ) do
    config
    |> new_client(http_opts)
    |> get(topic_url)
  end

  defp send_request(%{action: "DeleteTopic", params: %{topic_url: topic_url}}, config, http_opts) do
    config
    |> new_client(http_opts)
    |> delete(topic_url)
  end

  defp send_request(%{action: "ListTopics", headers: headers}, config, http_opts) do
    config
    |> new_client(http_opts)
    |> get("/topics", headers: headers)
  end

  defp send_request(
         %{
           action: "Subscribe",
           params: %{topic_url: topic_url, subscription_name: subscription_name} = params
         },
         config,
         http_opts
       ) do
    body = Xml.generate_subscription(params)

    config
    |> new_client(http_opts)
    |> put("#{subscription_url(topic_url, subscription_name)}", body)
  end

  defp send_request(
         %{
           action: "SetSubscriptionAttributes",
           params: %{topic_url: topic_url, subscription_name: subscription_name} = params
         },
         config,
         http_opts
       ) do
    body = Xml.generate_subscription(params)

    config
    |> new_client(http_opts)
    |> put("#{subscription_url(topic_url, subscription_name)}", body,
      query: [{"metaoverride", true}]
    )
  end

  defp send_request(
         %{
           action: "GetSubscriptionAttributes",
           params: %{topic_url: topic_url, subscription_name: subscription_name}
         },
         config,
         http_opts
       ) do
    config
    |> new_client(http_opts)
    |> get("#{subscription_url(topic_url, subscription_name)}")
  end

  defp send_request(
         %{
           action: "Unsubscribe",
           params: %{topic_url: topic_url, subscription_name: subscription_name}
         },
         config,
         http_opts
       ) do
    config
    |> new_client(http_opts)
    |> delete("#{subscription_url(topic_url, subscription_name)}")
  end

  defp send_request(
         %{action: "ListSubscriptions", headers: headers, params: %{topic_url: topic_url}},
         config,
         http_opts
       ) do
    config
    |> new_client(http_opts)
    |> get("#{topic_url}/subscriptions", headers: headers)
  end

  defp send_request(
         %{action: "PublishMessage", params: %{topic_url: topic_url} = params},
         config,
         http_opts
       ) do
    body = Xml.generate_topic_message(params)

    config
    |> new_client(http_opts)
    |> post("#{topic_url}/messages", body)
  end

  defp new_client(config, opts) do
    middleware = [
      {Tesla.Middleware.BaseUrl, config.host},
      {Tesla.Middleware.Retry, delay: 500, max_retries: 10, should_retry: &match_should_retry?/1},
      {ExAliyun.MNS.Http.Middleware, config}
    ]

    timeout = Keyword.get(opts, :timeout, 15_000)

    adapter = {
      Tesla.Adapter.Finch,
      name: Application.http_name(), receive_timeout: timeout
    }

    Tesla.client(middleware, adapter)
  end

  defp put(client, url, body, opts \\ []) do
    Tesla.put(client, url, body, opts)
  end

  defp get(client, url, opts \\ []) do
    Tesla.get(client, url, opts)
  end

  defp delete(client, url, opts \\ []) do
    Tesla.delete(client, url, opts)
  end

  defp post(client, url, body, opts \\ []) do
    Tesla.post(client, url, body, opts)
  end

  defp parse_response(env, action) do
    Parser.parse(env, action)
  end

  defp format_query(query) do
    format_query(query, [])
  end

  defp format_query([], result) do
    result
  end

  defp format_query([{_key, nil} | other], result) do
    format_query(other, result)
  end

  defp format_query([{key, value} | other], result) do
    format_query(other, [{key, value} | result])
  end

  defp subscription_url(topic_url, subscription_name) do
    "#{topic_url}/subscriptions/#{subscription_name}"
  end

  defp match_should_retry?({:error, :closed}), do: true
  defp match_should_retry?({:error, :timeout}), do: true
  defp match_should_retry?({:error, "socket closed"}), do: true
  defp match_should_retry?({:error, "timeout"}), do: true

  defp match_should_retry?({:error, error}) when is_atom(error) do
    Logger.error(fn ->
      "ExAliyunMNS occurs an unknown error: #{inspect(error)}, will retry it."
    end)

    true
  end

  defp match_should_retry?(_), do: false
end