raven_integration_maria_db

defmodule Integrations.MariaDb do
  @moduledoc """
  MariaDB / MySQL server monitor.

  Collection only — see `Integrations.MariaDb.Display` (same package) for
  the dashboard panel. `Display.BundledDefault` auto-hooks it whenever this
  monitor starts, same as a single-module package would; a release without
  `raven_web` simply never compiles the display half and runs this monitor
  headless.

  Connects via the MySQL wire protocol (v10) over raw TCP — no external library
  required. Authenticates with `mysql_native_password`, then collects connection,
  query throughput, InnoDB buffer pool, and replication metrics from
  `SHOW GLOBAL STATUS` and `SHOW VARIABLES`.

  ## Params

    * `:host`                — MariaDB hostname or IP. Required.
    * `:port`                — Port number. Defaults to `3306`.
    * `:user`                — MySQL username. Required.
    * `:password`            — Password. Blank means no password.
    * `:database`            — Database to connect to (optional but recommended).
    * `:timeout_ms`          — TCP connect and query timeout. Defaults to `5000`.
    * `:latency_degraded_ms` — Auth round-trip threshold for `:degraded`. Defaults to `500`.

  ## Health signal

    * `:up`       — Connected, authenticated, metrics within thresholds.
    * `:degraded` — Connection utilisation > 80 %, replication lag > 30 s,
                    or auth latency above threshold.
    * `:down`     — Connection refused, timeout, or auth failure.

  ## Metrics

    * `latency_ms`              — Time to TCP connect + authenticate
    * `threads_connected`       — Current connected client count
    * `connection_utilization`  — threads_connected / max_connections (0.0–1.0)
    * `queries_rate`            — Queries per second (rolling delta)
    * `slow_queries_rate`       — Slow queries per second (rolling delta)
    * `buffer_pool_hit_ratio`   — InnoDB buffer pool hit ratio percentage
    * `replication_lag_seconds` — Seconds_Behind_Master (replica only)
  """

  use CodeNameRaven.Monitor

  @default_port              3306
  @default_timeout_ms        5_000
  @default_lat_degraded_ms   500
  @default_conn_util_warn    0.8
  @default_repl_lag_warn_sec 30

  # CLIENT_LONG_PASSWORD | CLIENT_LONG_FLAG | CLIENT_CONNECT_WITH_DB |
  # CLIENT_PROTOCOL_41  | CLIENT_TRANSACTIONS | CLIENT_SECURE_CONNECTION |
  # CLIENT_PLUGIN_AUTH
  # = 0x1 | 0x4 | 0x8 | 0x200 | 0x2000 | 0x8000 | 0x80000 = 565_773
  @client_caps 565_773

  @impl true
  def params_template do
    %{
      host:     "mariadb.example.com",
      port:     "3306",
      user:     "root",
      password: "",
      database: ""
    }
  end

  @impl true
  def params_schema do
    [
      host:                [type: :string,          required: true,  doc: "MariaDB hostname or IP"],
      port:                [type: :non_neg_integer, default: 3306,   doc: "Port number"],
      user:                [type: :string,          required: true,  doc: "MySQL username"],
      password:            [type: :string,          required: false, doc: "Password (blank = no password)"],
      database:            [type: :string,          required: false, doc: "Database to connect to"],
      timeout_ms:          [type: :non_neg_integer, default: 5_000,  doc: "TCP connect and query timeout in milliseconds"],
      latency_degraded_ms: [type: :non_neg_integer, default: 500,    doc: "Auth latency threshold for degraded"]
    ]
  end

  @impl true
  def target_uri(params) do
    host = get_param(params, :host)
    port = parse_int(get_param(params, :port), @default_port)
    user = get_param(params, :user)
    db   = get_param(params, :database)
    cond do
      is_nil(host) or is_nil(user) -> :none
      db -> {:ok, "mysql://#{user}@#{host}:#{port}/#{db}"}
      true -> {:ok, "mysql://#{user}@#{host}:#{port}"}
    end
  end

  @impl true
  def identity_params(params) do
    %{
      host:     get_param(params, :host),
      port:     parse_int(get_param(params, :port), @default_port),
      user:     get_param(params, :user),
      database: get_param(params, :database)
    }
  end

  # ---------------------------------------------------------------------------
  # Collect
  # ---------------------------------------------------------------------------

  @impl true
  def collect(params, state) do
    host     = get_param(params, :host)
    port     = parse_int(get_param(params, :port), @default_port)
    user     = get_param(params, :user)
    password = get_param(params, :password) || ""
    database = get_param(params, :database)
    timeout  = parse_int(get_param(params, :timeout_ms), @default_timeout_ms)

    cond do
      is_nil(host) -> {:error, "missing required param :host", state}
      is_nil(user) -> {:error, "missing required param :user", state}
      true ->
        tcp_opts = [:binary, active: false, packet: :raw, send_timeout: timeout]

        case :gen_tcp.connect(to_charlist(host), port, tcp_opts, timeout) do
          {:ok, socket} ->
            result = run_checks(socket, user, password, database, timeout, state)
            :gen_tcp.close(socket)
            case result do
              {:ok, data, new_state} -> {:ok, data, new_state}
              {:error, reason}       -> {:error, reason, state}
            end

          {:error, :econnrefused} -> {:error, "connection refused on port #{port}", state}
          {:error, :timeout}      -> {:error, "connection timed out after #{timeout}ms", state}
          {:error, :nxdomain}     -> {:error, "hostname not found: #{host}", state}
          {:error, reason}        -> {:error, inspect(reason), state}
        end
    end
  end

  # ---------------------------------------------------------------------------
  # Healthy?
  # ---------------------------------------------------------------------------

  @impl true
  def healthy?(%{assertions: [_ | _], assertion_status: status}), do: status

  def healthy?(%{role_metrics: %{connection_utilization: cu}})
      when is_float(cu) and cu >= @default_conn_util_warn,
      do: :degraded

  def healthy?(%{replication: %{lag_seconds: lag}})
      when is_integer(lag) and lag >= @default_repl_lag_warn_sec,
      do: :degraded

  def healthy?(%{latency_ms: lat})
      when is_integer(lat) and lat >= @default_lat_degraded_ms,
      do: :degraded

  def healthy?(_result), do: :up

  # ---------------------------------------------------------------------------
  # Metrics
  # ---------------------------------------------------------------------------

  @impl true
  def metrics(result) do
    role = result[:role_metrics] || %{}
    db   = result[:db_metrics]   || %{}
    repl = result[:replication]  || %{}

    %{latency_ms: result.latency_ms}
    |> maybe_put(:threads_connected,       role[:threads_connected])
    |> maybe_put(:connection_utilization,  role[:connection_utilization])
    |> maybe_put(:queries_rate,            db[:queries_rate])
    |> maybe_put(:slow_queries_rate,       db[:slow_queries_rate])
    |> maybe_put(:buffer_pool_hit_ratio,   db[:buffer_pool_hit_ratio])
    |> maybe_put(:replication_lag_seconds, repl[:lag_seconds])
  end

  # ---------------------------------------------------------------------------
  # Connection + check orchestration
  # ---------------------------------------------------------------------------

  defp run_checks(socket, user, password, database, timeout, state) do
    t0 = System.monotonic_time(:millisecond)

    with {:ok, greeting}             <- read_packet(socket, timeout),
         {:ok, salt, server_version} <- parse_greeting(greeting),
         :ok                         <- send_auth(socket, user, password, database, salt),
         {:ok, auth_resp}            <- read_packet(socket, timeout),
         :ok                         <- check_ok_packet(auth_resp) do

      latency_ms = System.monotonic_time(:millisecond) - t0
      now        = DateTime.utc_now()
      elapsed    = elapsed_sec(state[:prev_collected_at], now)
      prev       = state[:prev_status] || %{}

      with {:ok, status_map} <- query_kv(socket, "SHOW GLOBAL STATUS", timeout),
           {:ok, var_map}    <- query_kv(socket,
             "SHOW VARIABLES WHERE Variable_name IN " <>
             "('max_connections','read_only','version')", timeout) do

        threads_connected = get_int(status_map, "Threads_connected")
        max_connections   = case get_int(var_map, "max_connections") do 0 -> 150; n -> n end
        is_replica        = (var_map["read_only"] || "") == "ON"
        version           = var_map["version"] || server_version

        conn_util = if max_connections > 0,
          do: Float.round(threads_connected / max_connections, 3),
          else: 0.0

        repl_lag =
          if is_replica do
            case query_slave_status(socket, timeout) do
              {:ok, lag} -> lag
              _          -> nil
            end
          else
            nil
          end

        result = %{
          latency_ms:  latency_ms,
          version:     version,
          is_replica:  is_replica,
          role_metrics: %{
            threads_connected:      threads_connected,
            max_connections:        max_connections,
            connection_utilization: conn_util
          },
          db_metrics: %{
            queries_rate:          delta_rate(status_map, prev, "Queries",     elapsed),
            slow_queries_rate:     delta_rate(status_map, prev, "Slow_queries", elapsed),
            buffer_pool_hit_ratio: buffer_hit_ratio(status_map, prev)
          },
          replication: %{
            lag_seconds: repl_lag
          }
        }

        new_state = %{prev_status: status_map, prev_collected_at: now}
        {:ok, result, new_state}
      end
    end
  end

  # ---------------------------------------------------------------------------
  # MySQL wire protocol — handshake
  # ---------------------------------------------------------------------------

  # Parse MySQL v10 server greeting; return {salt, server_version}.
  defp parse_greeting(<<10, after_version::binary>>) do
    [server_version, rest] = String.split(after_version, <<0>>, parts: 2)

    <<_thread_id::little-32,
      auth1::binary-8,
      0,
      _cap_lo::binary-2,
      _charset::8,
      _status::binary-2,
      _cap_hi::binary-2,
      auth_data_len::8,
      _reserved::binary-10,
      rest2::binary>> = rest

    # part 2 is max(13, auth_data_len - 8) bytes, last byte is null padding
    part2_len = max(13, auth_data_len - 8)
    <<auth2_padded::binary-size(^part2_len), _::binary>> = rest2
    auth2 = binary_part(auth2_padded, 0, part2_len - 1)

    {:ok, auth1 <> auth2, server_version}
  end
  defp parse_greeting(_), do: {:error, "unsupported MySQL protocol (expected v10)"}

  defp send_auth(socket, user, password, database, salt) do
    auth_resp = scramble(password, salt)
    db_part   = if database && database != "", do: database <> <<0>>, else: <<0>>

    payload =
      <<@client_caps::little-32,
        16_777_216::little-32,
        33::8,
        0::8*23>> <>
      user <> <<0>> <>
      <<byte_size(auth_resp)::8>> <>
      auth_resp <>
      db_part <>
      "mysql_native_password" <> <<0>>

    send_packet(socket, 1, payload)
  end

  # mysql_native_password: SHA1(pass) XOR SHA1(salt <> SHA1(SHA1(pass)))
  defp scramble("",  _salt), do: <<>>
  defp scramble(nil, _salt), do: <<>>
  defp scramble(password, salt) do
    s1 = :crypto.hash(:sha, password)
    s2 = :crypto.hash(:sha, s1)
    s3 = :crypto.hash(:sha, salt <> s2)
    :crypto.exor(s1, s3)
  end

  defp check_ok_packet(<<0x00, _::binary>>), do: :ok
  defp check_ok_packet(<<0xFE, _::binary>>), do: :ok
  defp check_ok_packet(<<0xFF, _code::little-16, "#", _state::binary-5, msg::binary>>),
    do: {:error, "MySQL auth error: #{msg}"}
  defp check_ok_packet(<<0xFF, _::binary>>),
    do: {:error, "MySQL auth failed"}
  defp check_ok_packet(other),
    do: {:error, "unexpected server response: #{inspect(binary_part(other, 0, min(20, byte_size(other))))}"}

  # ---------------------------------------------------------------------------
  # MySQL wire protocol — packets
  # ---------------------------------------------------------------------------

  defp read_packet(socket, timeout) do
    case :gen_tcp.recv(socket, 4, timeout) do
      {:ok, <<len::little-24, _seq::8>>} ->
        case :gen_tcp.recv(socket, len, timeout) do
          {:ok, payload} -> {:ok, payload}
          {:error, r}    -> {:error, "payload read failed: #{inspect(r)}"}
        end
      {:error, r} ->
        {:error, "packet header read failed: #{inspect(r)}"}
    end
  end

  defp send_packet(socket, seq, payload) do
    len = byte_size(payload)
    :gen_tcp.send(socket, <<len::little-24, seq::8>> <> payload)
  end

  # ---------------------------------------------------------------------------
  # MySQL wire protocol — query + text result set
  # ---------------------------------------------------------------------------

  # Send COM_QUERY and return rows as a list of column-name → value maps.
  defp query(socket, sql, timeout) do
    :ok = send_packet(socket, 0, <<0x03>> <> sql)
    read_result_set(socket, timeout)
  end

  # For SHOW STATUS / SHOW VARIABLES: two-column (Variable_name, Value) rows
  # collapsed into a flat %{"name" => "value"} map.
  defp query_kv(socket, sql, timeout) do
    case query(socket, sql, timeout) do
      {:ok, rows} ->
        map =
          Map.new(rows, fn row ->
            {row["Variable_name"] || row["variable_name"] || "",
             row["Value"]         || row["value"]         || ""}
          end)
        {:ok, map}
      err -> err
    end
  end

  defp query_slave_status(socket, timeout) do
    case query(socket, "SHOW SLAVE STATUS", timeout) do
      {:ok, [row | _]} ->
        lag = case row["Seconds_Behind_Master"] do
          nil    -> nil
          "NULL" -> nil
          v      -> parse_int_or_nil(v)
        end
        {:ok, lag}
      {:ok, []}   -> {:ok, nil}
      {:error, r} -> {:error, r}
    end
  end

  defp read_result_set(socket, timeout) do
    case read_packet(socket, timeout) do
      {:ok, <<0x00, _::binary>>} ->
        {:ok, []}

      {:ok, <<0xFF, _code::little-16, "#", _state::binary-5, msg::binary>>} ->
        {:error, "MySQL error: #{msg}"}

      {:ok, col_count_payload} ->
        col_count = decode_lenenc(col_count_payload)
        with {:ok, columns} <- read_column_defs(socket, timeout, col_count),
             :ok            <- skip_eof_or_ok(socket, timeout),
             {:ok, rows}    <- read_rows(socket, timeout, columns) do
          {:ok, rows}
        end

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp read_column_defs(_socket, _timeout, 0), do: {:ok, []}
  defp read_column_defs(socket, timeout, count) do
    Enum.reduce_while(1..count//1, {:ok, []}, fn _, {:ok, acc} ->
      case read_packet(socket, timeout) do
        {:ok, col_def} -> {:cont, {:ok, acc ++ [column_name(col_def)]}}
        {:error, r}    -> {:halt, {:error, r}}
      end
    end)
  end

  # Column definition packet: catalog, schema, table, org_table, name, org_name
  # Each field is a length-encoded string; we want the 5th (name).
  defp column_name(data) do
    {_, r1} = read_lenenc_str(data)
    {_, r2} = read_lenenc_str(r1)
    {_, r3} = read_lenenc_str(r2)
    {_, r4} = read_lenenc_str(r3)
    {name, _} = read_lenenc_str(r4)
    name || ""
  end

  defp skip_eof_or_ok(socket, timeout) do
    case read_packet(socket, timeout) do
      {:ok, _} -> :ok
      {:error, r} -> {:error, r}
    end
  end

  defp read_rows(socket, timeout, columns), do: read_rows_acc(socket, timeout, columns, [])

  defp read_rows_acc(socket, timeout, columns, acc) do
    case read_packet(socket, timeout) do
      {:ok, <<0xFE, _::binary>>} ->
        {:ok, Enum.reverse(acc)}

      {:ok, <<0xFF, _code::little-16, "#", _state::binary-5, msg::binary>>} ->
        {:error, "MySQL error: #{msg}"}

      {:ok, row_payload} ->
        {values, _} =
          Enum.reduce(columns, {[], row_payload}, fn _, {vals, rest} ->
            {val, remaining} = read_lenenc_str(rest)
            {vals ++ [val], remaining}
          end)
        row = Enum.zip(columns, values) |> Map.new()
        read_rows_acc(socket, timeout, columns, [row | acc])

      {:error, r} ->
        {:error, r}
    end
  end

  # ---------------------------------------------------------------------------
  # Length-encoded integers and strings
  # ---------------------------------------------------------------------------

  defp decode_lenenc(<<n, _::binary>>) when n < 251, do: n
  defp decode_lenenc(<<0xFC, n::little-16, _::binary>>), do: n
  defp decode_lenenc(<<0xFD, n::little-24, _::binary>>), do: n
  defp decode_lenenc(<<0xFE, n::little-64, _::binary>>), do: n
  defp decode_lenenc(_), do: 0

  defp read_lenenc_str(<<0xFB, rest::binary>>), do: {nil, rest}
  defp read_lenenc_str(<<len, rest::binary>>) when len < 251 do
    <<str::binary-size(^len), remaining::binary>> = rest
    {str, remaining}
  end
  defp read_lenenc_str(<<0xFC, len::little-16, rest::binary>>) do
    <<str::binary-size(^len), remaining::binary>> = rest
    {str, remaining}
  end
  defp read_lenenc_str(<<0xFD, len::little-24, rest::binary>>) do
    <<str::binary-size(^len), remaining::binary>> = rest
    {str, remaining}
  end
  defp read_lenenc_str(<<0xFE, len::little-64, rest::binary>>) do
    <<str::binary-size(^len), remaining::binary>> = rest
    {str, remaining}
  end
  defp read_lenenc_str(data), do: {"", data}

  # ---------------------------------------------------------------------------
  # Metric computation
  # ---------------------------------------------------------------------------

  defp get_int(map, key) do
    case Integer.parse(map[key] || "") do
      {n, _} -> n
      :error -> 0
    end
  end

  defp delta_rate(current, prev, key, elapsed)
       when is_number(elapsed) and elapsed > 0 do
    cur  = get_int(current, key)
    prv  = get_int(prev, key)
    if cur >= prv, do: Float.round((cur - prv) / elapsed, 2), else: nil
  end
  defp delta_rate(_, _, _, _), do: nil

  defp buffer_hit_ratio(current, prev) do
    reads    = get_int(current, "Innodb_buffer_pool_reads")
    requests = get_int(current, "Innodb_buffer_pool_read_requests")
    prev_rd  = get_int(prev,    "Innodb_buffer_pool_reads")
    prev_req = get_int(prev,    "Innodb_buffer_pool_read_requests")
    d_req = requests - prev_req
    d_rd  = reads    - prev_rd
    if d_req > 0, do: Float.round((d_req - d_rd) / d_req * 100, 1), else: nil
  end

  defp elapsed_sec(nil, _now), do: nil
  defp elapsed_sec(prev, now) do
    diff = DateTime.diff(now, prev, :millisecond)
    if diff > 0, do: diff / 1000, else: nil
  end

  defp parse_int_or_nil(v) do
    case Integer.parse(v || "") do
      {n, _} -> n
      :error -> nil
    end
  end

  # ---------------------------------------------------------------------------
  # Helpers
  # ---------------------------------------------------------------------------

  defp maybe_put(map, _key, nil), do: map
  defp maybe_put(map, key, val),  do: Map.put(map, key, val)

  defp get_param(params, key) when is_atom(key) do
    v = params[key] || params[to_string(key)]
    if is_binary(v) and String.trim(v) == "", do: nil, else: v
  end

  defp parse_int(nil, default), do: default
  defp parse_int(v, _) when is_integer(v), do: v
  defp parse_int(v, default) when is_binary(v) do
    case Integer.parse(v) do
      {n, _} -> n
      :error -> default
    end
  end
  defp parse_int(_, default), do: default
end