defmodule Gelfx do
  @moduledoc """
  A logger backend for Elixir applications using `Logger` and Graylog based on GELF (Graylog extended logging format).

  ## Usage
  Add Gelfx to your application by adding `{:gelfx, "~> #{Mix.Project.config()[:version]}"}` to your list of dependencies in `mix.exs`:
  ```elixir
  def deps do
    [
      # ...
      {:gelfx, "~> #{Mix.Project.config()[:version]}"}
    ]
  end
  ```
  And adding it to your `:logger` configuration in `config.exs`:
  ```elixir
  config :logger,
    backends: [
      :console,
      Gelfx
    ]
  ```
  Since Elixir 1.15 Logger backends live in the `:logger_backends` package, which Gelfx depends on.
  Backends can also be added at runtime using `LoggerBackends.add/1`:
  ```elixir
  LoggerBackends.add(Gelfx)
  ```
  Since GELF relies on json to encode the payload Gelfx uses the `JSON` module included in Elixir (requires Elixir 1.18 or later).

  ## Options
  Besides `:level`, `:format` and `:metadata`, which are advised by [Logger](https://hexdocs.pm/logger/Logger.html#module-custom-backends) Gelfx supports:
  - `:host` - hostname of the server running the GELF endpoint, defaults to `localhost`
  - `:port` - port on which the graylog server runs the respective GELF input, defaults to `12201`
  - `:protocol` - either `:tcp`, `:udp`, or `:http`.  Defaults to `:udp`.
  - `:connection_timeout` - sets the timeout in ms after which the `:tcp` connect timeouts, defaults to 5s.
  - `:compression` - either `:gzip` or `:zlib` can be set and will be used for package compression when UDP or HTTP (only gzip) is used as protocol
  - `:format` - defaults to `"$message"`
  - `:hostname` - used as source field in the GELF message, defaults to the hostname returned by `:inet.gethostname()`
  - `:utc_log` - this option should not be configured directly. But rather by setting the `:utc_log` option in the `Logger` config. Should the Logger config change after the Gelfx backend is initialized the option has to be reconfigured.

  ### HTTP
  When using the HTTP protocol the url is build using the scheme defined in the graylog [documentation](https://docs.graylog.org/en/3.1/pages/sending_data.html#gelf-via-http).

  Setting the host to `localhost` and port to `12201` sends log entries to `http://localhost:12201/gelf`.

  ## Message Format
  The GELF message format version implemented by this library is 1.1, the docs can be found [here](http://docs.graylog.org/en/3.0/pages/gelf.html).

  Messages can include a `short_message` and a `full_message`, Gelfx will use the first line of each log message for the `short_message` and will place the whole message in the `full_message` field.

  Metadata will be included in the message using the _additional field_ syntax.
  The Keys of the metadata entries have to match `^\\_?[\\w\\.\\-]*$`, keys missing an leading underscore are automatically prepended with one.
  Key collisions are __NOT__ prevented by Gelfx, additionally the keys `id` and `_id` are automatically omitted due to the GELF specification.

  ## Custom formatting
  You can use your own log formatter in the same way you would define one for the Elixir default Logger.
  More information can be found [here](https://hexdocs.pm/logger/Logger.html#module-custom-formatting) in the Logger documentation.

  ```elixir
    config :logger, Gelfx,
      format: {MyCustomFormatter, :format}
  ```

  ## Levels
  Graylog relies on the syslog definitions for logging levels.
  Gelfx maps the Elixir levels as follows:

  | Elixir | Syslog | GELF - Selector | Note |
  |-|-|-|-|
  | `:emergency` | Emergency | 0 | |
  | `:alert` | Alert | 1 | |
  | `:critical` | Critical | 2 | |
  | `:error` | Error | 3 | |
  | `:warning` | Warning | 4 | |
  | `:warn` | -- | -- | Same as `:warning` was soft-deprecated in elixir 1.11 |
  | `:notice` | Notice | 5 | |
  | `:info` | Informational | 6 | |
  | `:debug` | Debug | 7 | |
  """

  # Copyright 2019 Hans Bernhard Goedeke
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.

  @behaviour :gen_event

  alias Gelfx.LogEntry
  alias Logger.Formatter

  require Logger

  defstruct [
    :compression,
    :conn,
    :connection_timeout,
    :format,
    :host,
    :hostname,
    :level,
    :metadata,
    :port,
    :protocol,
    :utc_log
  ]

  @default_conf [
    connection_timeout: 5_000,
    format: "$message",
    host: "localhost",
    metadata: [],
    port: 12_201,
    protocol: :udp
  ]

  # 2 magic bytes + 8 Msg ID + 1 seq number + 1 seq count
  @chunk_header_bytes 12

  @impl true
  def init(__MODULE__) do
    init({__MODULE__, []})
  end

  def init({__MODULE__, options}) do
    {:ok, config(options, %__MODULE__{})}
  end

  defp config(options, %__MODULE__{} = state) do
    config =
      default_conf()
      |> Keyword.merge(Application.get_env(:logger, __MODULE__, []))
      |> Keyword.merge(options)

    Application.put_env(:logger, __MODULE__, config)

    state = %__MODULE__{
      state
      | compression: Keyword.get(config, :compression),
        connection_timeout: Keyword.get(config, :connection_timeout),
        format: Formatter.compile(Keyword.get(config, :format)),
        host: Keyword.get(config, :host),
        hostname: Keyword.get(config, :hostname),
        level: Keyword.get(config, :level),
        metadata: Keyword.get(config, :metadata),
        port: Keyword.get(config, :port),
        protocol: Keyword.get(config, :protocol),
        utc_log: Keyword.get(config, :utc_log)
    }

    case :ets.info(__MODULE__, :size) do
      :undefined ->
        :ets.new(__MODULE__, [:public, :duplicate_bag, :named_table])

      0 ->
        :ok

      _ ->
        send(self(), {__MODULE__, :flush_buffer})
    end

    case spawn_conn(state) do
      {:error, reason} ->
        Logger.error(["Cloud not establish connection on init ", inspect(reason)])
        %{state | conn: :error}

      conn ->
        %{state | conn: conn}
    end
  end

  @impl true
  def handle_call({:configure, options}, state) do
    {:ok, :ok, config(options, state)}
  end

  def handle_call(_, state) do
    {:ok, :ok, state}
  end

  @impl true
  def handle_event(event, %{conn: :error} = state) do
    case handle_info(:retry, state) do
      {:ok, state} ->
        handle_event(event, state)

      error ->
        error
    end
  end

  def handle_event(_, %{conn: {:retry, :discard}} = state) do
    {:ok, state}
  end

  def handle_event(
        {level, group_leader, {Logger, _message, _timestamp, _metadata}} = event,
        %{conn: {:retry, _mode}} = state
      ) do
    if meet_level?(level, state.level) and node(group_leader) == node() do
      :ets.insert(__MODULE__, {:erlang.monotonic_time(), event})
    end

    {:ok, state}
  end

  def handle_event({level, group_leader, {Logger, _message, _timestamp, _metadata}} = event, state) do
    if meet_level?(level, state.level) and node(group_leader) == node() do
      event
      |> LogEntry.from_event(state)
      |> encode()
      |> case do
        {:ok, json} ->
          submit(json, state)

        error ->
          error
      end
    end

    {:ok, state}
  end

  def handle_event(:flush, state) do
    flush(state.conn)
    {:ok, state}
  end

  def handle_event(_event, state) do
    {:ok, state}
  end

  @impl true
  def handle_info({__MODULE__, msg}, state) do
    handle_info(msg, state)
  end

  def handle_info({:tcp_error, socket, reason}, state) do
    case state do
      %{conn: {:tcp, ^socket}} ->
        Logger.warning(["TCP connection error ", inspect(reason)])
        handle_info(:retry, close_conn(state))

      _ ->
        {:ok, state}
    end
  end

  def handle_info({:tcp_closed, socket}, state) do
    case state do
      %{conn: {:tcp, ^socket}} ->
        Logger.warning("TCP connection closed")
        handle_info(:retry, close_conn(state))

      _ ->
        {:ok, state}
    end
  end

  def handle_info(:flush_buffer, %{conn: {prot, _}} = state) when prot in [:tcp, :udp] do
    case :ets.first(__MODULE__) do
      :"$end_of_table" ->
        {:ok, state}

      key ->
        send(self(), :flush_buffer)

        # this has to be `Enum.reduce` since each event is stored using the
        # erlang monotonic time, thus duplicates are possible.
        Enum.reduce(
          :ets.take(__MODULE__, key),
          {:ok, state},
          fn {_, event}, acc ->
            case acc do
              {:ok, state} ->
                handle_event(event, state)

              error ->
                error
            end
          end
        )
    end
  end

  def handle_info(:retry, state) do
    if state.conn in [:error, nil], do: Logger.debug("Connection error starting retries")

    case spawn_conn(state) do
      {:error, _reason} ->
        Process.send_after(self(), :retry, state.connection_timeout)

        mode =
          if discard?() do
            :discard
          else
            :store
          end

        {:ok, %{state | conn: {:retry, mode}}}

      conn ->
        Logger.info("Connection established flushing buffer")
        handle_info(:flush_buffer, %{state | conn: conn})
    end
  end

  def handle_info(_msg, state) do
    {:ok, state}
  end

  @impl true
  def code_change(_old_vsn, state, _extra) do
    {:ok, state}
  end

  @impl true
  def terminate(:swap, state) do
    state = close_conn(state)

    [
      compression: state.compression,
      format: state.format,
      host: state.host,
      level: state.level,
      metadata: state.metadata,
      port: state.port,
      protocol: state.protocol
    ]
  end

  def terminate(_reason, state) do
    close_conn(state)
    :ets.delete(__MODULE__)
  end

  @doc """
  Encodes the given `LogEntry` using the `JSON` module included in Elixir
  """
  def encode(log_entry) when is_map(log_entry) do
    {:ok, JSON.encode!(log_entry)}
  rescue
    error -> {:error, error}
  end

  @compile {:inline, meet_level?: 2}
  @doc false
  def meet_level?(_lvl, nil), do: true

  # events dispatched to legacy backends still carry :warn instead of :warning
  def meet_level?(:warn, min), do: meet_level?(:warning, min)

  def meet_level?(lvl, :warn), do: meet_level?(lvl, :warning)

  def meet_level?(lvl, min) do
    Logger.compare_levels(lvl, min) != :lt
  end

  @doc """
  Spawns a `:gen_udp` / `:gen_tcp` connection based on the configuration
  """
  def spawn_conn(%__MODULE__{conn: {prot, _socket} = conn} = state) when prot in [:udp, :tcp] do
    flush(conn)

    state
    |> close_conn()
    |> spawn_conn()
  end

  def spawn_conn(%__MODULE__{protocol: protocol, host: host, port: port, connection_timeout: timeout}) do
    spawn_conn(protocol, host, port, timeout)
  end

  def spawn_conn(protocol, host, port, timeout) when is_binary(host) do
    spawn_conn(protocol, String.to_charlist(host), port, timeout)
  end

  def spawn_conn(:tcp, host, port, timeout) do
    case :gen_tcp.connect(host, port, [:binary, active: true], timeout) do
      {:ok, socket} -> {:tcp, socket}
      error -> error
    end
  end

  def spawn_conn(:udp, host, port, _timeout) do
    case :gen_udp.open(0, [:binary, active: true]) do
      {:ok, socket} ->
        buffer =
          case :inet.getopts(socket, [:buffer]) do
            {:ok, [buffer: buffer]} -> buffer
            _ -> 8192
          end

        {:udp, {socket, host, port, buffer}}

      error ->
        error
    end
  end

  def spawn_conn(:http, host, port, _timeout) do
    :inets.start()

    {:http, List.flatten([~c"http://", host, ?:, Integer.to_charlist(port) | ~c"/gelf"])}
  end

  @doc false
  def close_conn(%__MODULE__{conn: connection} = state) do
    case close_conn(connection) do
      :ok -> %{state | conn: nil}
      :error -> state
    end
  end

  def close_conn({:tcp, socket}) do
    :gen_tcp.close(socket)
  end

  def close_conn({:udp, {socket, _host, _post, _buffer}}) do
    :gen_udp.close(socket)
  end

  def close_conn({:http, _}) do
    :ok
  end

  def close_conn(_) do
    :error
  end

  @doc """
  Sends the given payload over the connection.

  In case an TCP connection is used the `0x00` delimiter required by gelf is added.

  Should the used connection use UDP the payload is compressed using the configured compression, in case the given payload exceeds the chunk threshold it is chunked.
  """
  def submit(payload, %__MODULE__{conn: conn, compression: comp}) do
    submit(payload, conn, comp)
  end

  # TCP
  def submit(payload, {:tcp, socket}, _compression) do
    case :gen_tcp.send(socket, payload <> <<0>>) do
      :ok -> :ok
      {:error, :closed} -> :retry
      _ -> :error
    end
  end

  # UDP
  def submit(payload, {:udp, _} = conn, :gzip) do
    payload
    |> :zlib.gzip()
    |> submit(conn, nil)
  end

  def submit(payload, {:udp, _} = conn, :zlib) do
    payload
    |> :zlib.compress()
    |> submit(conn, nil)
  end

  def submit(payload, {:udp, {socket, host, port, chunk_threshold}}, _comp) when byte_size(payload) <= chunk_threshold do
    case :gen_udp.send(socket, host, port, payload) do
      :ok -> :ok
      _ -> :error
    end
  end

  def submit(payload, {:udp, {_socket, _host, _port, chunk_threshold}} = conn, comp) do
    chunks = chunk(payload, chunk_threshold - @chunk_header_bytes)
    chunk_count = length(chunks)
    msg_id = message_id()

    for {seq_nr, chunck} <- chunks do
      submit(
        <<0x1E, 0x0F, msg_id::bytes-size(8), seq_nr::8, chunk_count::8, chunck::binary>>,
        conn,
        comp
      )
    end
  end

  # HTTP
  def submit(payload, {:http, url}, :gzip) do
    :httpc.request(
      :post,
      {url, [{~c"Content-Encoding", ~c"gzip"}], ~c"application/json", :zlib.gzip(payload)},
      [],
      []
    )

    :ok
  end

  def submit(payload, {:http, url}, _comp) do
    :httpc.request(:post, {url, [], ~c"application/json", payload}, [], [])

    :ok
  end

  defp flush({:tcp, socket}) do
    flush(socket)
  end

  defp flush({:udp, {socket, _host, _post, _buffer}}) do
    flush(socket)
  end

  defp flush(socket) when is_port(socket) do
    case :inet.getstat(socket, [:send_pend]) do
      {:ok, [send_pend: 0]} ->
        :ok

      {:ok, _} ->
        Process.sleep(10)
        flush(socket)

      _ ->
        :error
    end
  end

  defp flush({:http, _}) do
    :ok
  end

  defp flush(_) do
    :error
  end

  defp chunk(binary, chunk_length, sequence_number \\ 0) do
    case binary do
      <<chunk::bytes-size(^chunk_length), rest::binary>> ->
        [{sequence_number, chunk} | chunk(rest, chunk_length, sequence_number + 1)]

      _ ->
        [{sequence_number, binary}]
    end
  end

  defp message_id do
    monotonic_timestamp = :erlang.monotonic_time()

    os_timestamp = :calendar.time_to_seconds(:os.timestamp())

    <<os_timestamp::integer-32, monotonic_timestamp::integer-32>>
  end

  defp discard? do
    :ets.info(__MODULE__, :size) >= Application.get_env(:logger, :discard_threshold, 500) * 10
  end

  defp hostname do
    {:ok, hostname} = :inet.gethostname()
    to_string(hostname)
  end

  defp default_conf do
    @default_conf
    |> Keyword.put(:hostname, hostname())
    |> Keyword.put(:utc_log, Application.get_env(:logger, :utc_log, false))
  end
end