nebula_graph_ex

defmodule NebulaGraphEx.Graph do
  @moduledoc """
  Public API for querying NebulaGraph.

  ## The recommended pattern — define a graph module

  Borrowing from Ecto's Repo pattern, define a graph module in your application:

      # lib/my_app/graph.ex
      defmodule MyApp.Graph do
        use NebulaGraphEx.Graph, otp_app: :my_app
      end

  Put connection options in config:

      # config/config.exs
      config :my_app, MyApp.Graph,
        hostname: "localhost",
        port: 9669,
        username: "root",
        space: "my_graph",
        pool_size: 10

  Override secrets at runtime (recommended for passwords):

      # config/runtime.exs
      import Config

      config :my_app, MyApp.Graph,
        hostname: System.fetch_env!("NEBULA_HOST"),
        password: fn -> System.fetch_env!("NEBULA_PASS") end

  Add to your supervision tree — that's it:

      # lib/my_app/application.ex
      def start(_type, _args) do
        children = [MyApp.Graph]
        Supervisor.start_link(children, strategy: :one_for_one)
      end

  Now query without ever referencing a PID or pool name:

      MyApp.Graph.query("MATCH (v:Player) RETURN v.name, v.age LIMIT 5")
      MyApp.Graph.query!("RETURN 1+1 AS n") |> ResultSet.first!() |> Record.get!("n")

  ## Direct pool usage (without `use`)

  When you need multiple pools, or want full control, start `NebulaGraphEx.Graph`
  directly in your supervision tree:

      children = [
        {NebulaGraphEx.Graph,
         name: MyApp.Graph,
         hostname: "localhost",
         port: 9669,
         space: "my_graph",
         pool_size: 10}
      ]

  Then pass the name (or PID) to every call:

      NebulaGraphEx.Graph.query(MyApp.Graph, "MATCH (v:Player) RETURN v LIMIT 5")

  ## Parameterised queries

  Always use parameters for user-supplied values — never interpolate into the
  statement string.

      MyApp.Graph.query(
        "MATCH (v:Player{name: $name}) RETURN v.age AS age",
        %{"name" => "Tim Duncan"}
      )

  ## Per-query options

  Any pool option can be overridden for a single call:

      MyApp.Graph.query(stmt, %{},
        space: "other_space",
        timeout: 30_000,
        decode_mapper: &NebulaGraphEx.Record.to_map/1
      )

  ## Bang variants

  `query!/3` raises `NebulaGraphEx.Error` on failure instead of returning
  `{:error, error}`.

  ## Query logging

  Enable structured query logging to see what is being sent to NebulaGraph,
  where it was called from, and how long it took.

  Turn on logging for all queries from a pool via config:

      config :my_app, MyApp.Graph,
        log: true,
        log_level: :debug   # optional, :debug is the default

  Enable logging for a single call:

      MyApp.Graph.query("MATCH (v:Player) RETURN v.name LIMIT 5", %{}, log: true)

  Log output includes the caller location, status, row count, db time, queue
  time, the nGQL statement, and any parameters:

      ↳ MyApp.Players.find/1, at: lib/my_app/players.ex:17
      QUERY OK 3 rows db=0.6ms queue=0.1ms
      MATCH (v:Player) RETURN v.name LIMIT 3 []

  To surface only slow queries without logging everything, use
  `:slow_query_threshold` (in milliseconds):

      config :my_app, MyApp.Graph,
        slow_query_threshold: 100

  See `NebulaGraphEx.Options` for the full option reference including
  `:log`, `:log_level`, and `:slow_query_threshold`.

  ## Connection status

  Use `status/1` or `status/2` when you want a health check for a pool. It
  verifies that the pool process is alive and, by default, executes `RETURN 1`
  to confirm the connection can run a query.

      {:ok, status} = MyApp.Graph.status()
      status.connected?
      #=> true

  See `NebulaGraphEx.Options` for the full option reference.
  """

  require Logger

  alias NebulaGraphEx.Connection
  alias NebulaGraphEx.Error
  alias NebulaGraphEx.Options
  alias NebulaGraphEx.Query
  alias NebulaGraphEx.Result
  alias NebulaGraphEx.ResultSet
  alias NebulaGraphEx.Telemetry

  @doc """
  Injects a graph module into the calling module.

  ## Options

  * `:otp_app` — required. The OTP application atom used to look up config via
    `Application.get_env(otp_app, __MODULE__, [])`.

  ## Generated API

  When you `use NebulaGraphEx.Graph, otp_app: :my_app`, the following functions
  are injected into your module, all scoped to the pool that the module itself
  manages:

  * `start_link/1` — starts the pool (used by your supervision tree).
  * `child_spec/1` — makes the module a valid child spec entry.
  * `query/1,2,3` — `query(statement, params \\\\ %{}, opts \\\\ [])`.
  * `query!/1,2,3` — raises on failure.
  * `map/2,3,4` — `map(statement, params \\\\ %{}, opts \\\\ [], fun)`.
  * `status/0,1` — checks whether the pool is alive and can run a simple query.
  * `use_space/1,2` — `use_space(space, opts \\\\ [])`.

  All functions are overridable so you can add instrumentation or defaults.
  """
  defmacro __using__(opts) do
    otp_app = Keyword.fetch!(opts, :otp_app)

    quote do
      @otp_app unquote(otp_app)

      @doc """
      Starts the `#{__MODULE__}` connection pool.

      Reads options from `Application.get_env(#{@otp_app}, #{__MODULE__}, [])`,
      then merges any `runtime_opts` passed here. The pool is registered under
      the module name so you never need to pass a PID to query functions.
      """
      def start_link(runtime_opts \\ []) do
        opts =
          Application.get_env(@otp_app, __MODULE__, [])
          |> Keyword.merge(runtime_opts)
          |> Keyword.put_new(:name, __MODULE__)

        NebulaGraphEx.Graph.start_link(opts)
      end

      @doc false
      def child_spec(opts) do
        %{
          id: __MODULE__,
          start: {__MODULE__, :start_link, [opts]},
          type: :worker
        }
      end

      @doc """
      Executes a nGQL statement. See `NebulaGraphEx.Graph.query/4`.

      Pool-level `:log`, `:log_level`, and `:slow_query_threshold` options are
      read from application config and merged into every call. Per-call opts
      take precedence.
      """
      def query(statement, params \\ %{}, opts \\ []) do
        config = Application.get_env(@otp_app, __MODULE__, [])
        log_opts = Keyword.take(config, [:log, :log_level, :slow_query_threshold])
        # per-call opts win over pool config
        opts = Keyword.merge(log_opts, opts)

        # Capture caller location now (before crossing into library internals)
        # so the log line points at user code, not library internals.
        opts =
          if NebulaGraphEx.Graph.logging_enabled?(opts) do
            Keyword.put_new(
              opts,
              :nebula_caller,
              {:erlang.process_info(self(), :current_stacktrace), __MODULE__}
            )
          else
            opts
          end

        NebulaGraphEx.Graph.query(__MODULE__, statement, params, opts)
      end

      @doc """
      Like `query/3` but raises `NebulaGraphEx.Error` on failure.
      """
      def query!(statement, params \\ %{}, opts \\ []) do
        case query(statement, params, opts) do
          {:ok, rs} -> rs
          {:error, err} -> raise err
        end
      end

      @doc """
      Executes a nGQL statement and maps `fun` over every result row.
      See `NebulaGraphEx.Graph.map/5`.
      """
      def map(statement, params \\ %{}, opts \\ [], fun) do
        case query(statement, params, opts) do
          {:ok, rs} -> {:ok, ResultSet.map(rs, fun)}
          {:error, _} = err -> err
        end
      end

      @doc """
      Checks whether the pool process is alive and can execute a simple query.
      See `NebulaGraphEx.Graph.status/2`.
      """
      def status(opts \\ []) do
        NebulaGraphEx.Graph.status(__MODULE__, opts)
      end

      @doc """
      Issues `USE <space>` on the pool. See `NebulaGraphEx.Graph.use_space/3`.
      """
      def use_space(space, opts \\ []) do
        NebulaGraphEx.Graph.use_space(__MODULE__, space, opts)
      end

      defoverridable start_link: 1,
                     child_spec: 1,
                     query: 3,
                     query!: 3,
                     map: 4,
                     status: 1,
                     use_space: 2
    end
  end

  @doc """
  Starts a supervised connection pool and links it to the calling process.

  Typically called via a module that uses `NebulaGraphEx.Graph`. When using the
  pool directly, pass a `:name` so callers can reference it by atom:

      children = [
        {NebulaGraphEx.Graph,
         name: MyApp.Graph,
         hostname: System.fetch_env!("NEBULA_HOST"),
         password: fn -> System.fetch_env!("NEBULA_PASS") end,
         space: "my_graph",
         pool_size: 20}
      ]

  See `NebulaGraphEx.Options` for the full option reference.
  """
  @spec start_link(keyword()) :: {:ok, pid()} | {:error, term()}
  def start_link(opts) do
    opts = Options.validate_pool!(opts)
    DBConnection.start_link(Connection, opts)
  end

  @doc false
  def child_spec(opts) do
    %{
      id: Keyword.get(opts, :name, __MODULE__),
      start: {__MODULE__, :start_link, [opts]},
      type: :worker
    }
  end

  @doc """
  Executes a nGQL statement and returns `{:ok, %ResultSet{}}` or
  `{:error, %NebulaGraphEx.Error{}}`.

  ## Arguments

  * `conn` — pool name or PID from `start_link/1`. When using the `use` macro,
    this is the module itself and is passed automatically.
  * `statement` — nGQL string.
  * `params` — map of parameter bindings. Default: `%{}`.
  * `opts` — per-query option overrides (see `NebulaGraphEx.Options`). Default: `[]`.

  ## Examples

      {:ok, rs} = NebulaGraphEx.Graph.query(MyApp.Graph, "RETURN 1+1 AS sum")
      rs |> NebulaGraphEx.ResultSet.first!() |> NebulaGraphEx.Record.get!("sum")
      #=> 2

      {:error, %NebulaGraphEx.Error{code: :e_syntax_error}} =
        NebulaGraphEx.Graph.query(MyApp.Graph, "THIS IS NOT VALID nGQL")
  """
  @spec query(DBConnection.conn(), String.t(), map(), keyword()) ::
          {:ok, ResultSet.t()} | {:error, Error.t()}
  def query(conn, statement, params \\ %{}, opts \\ []) do
    telemetry_prefix = Keyword.get(opts, :telemetry_prefix, [:nebula_graph_ex, :query])
    meta = %{statement: statement, params: params, opts: opts}

    queue_start = System.monotonic_time(:microsecond)

    result =
      Telemetry.span(telemetry_prefix, meta, fn ->
        q = Query.new(statement, params, opts)
        decode_mapper = Keyword.get(opts, :decode_mapper)
        query_start = System.monotonic_time(:microsecond)

        outcome =
          case DBConnection.execute(conn, q, [], build_db_opts(opts)) do
            {:ok, _query, %Result{} = r} -> {:ok, ResultSet.from_result(r, decode_mapper)}
            {:error, %Error{} = err} -> {:error, err}
            {:error, reason} -> {:error, Error.client_error(inspect(reason))}
          end

        query_us = System.monotonic_time(:microsecond) - query_start
        queue_us = query_start - queue_start
        maybe_log(statement, params, outcome, queue_us, query_us, opts)
        outcome
      end)

    result
  end

  @doc """
  Like `query/4` but raises `NebulaGraphEx.Error` on failure.

  ## Example

      NebulaGraphEx.Graph.query!(MyApp.Graph, "MATCH (v:Player) RETURN count(v) AS n")
      |> NebulaGraphEx.ResultSet.first!()
      |> NebulaGraphEx.Record.get!("n")
      #=> 51
  """
  @spec query!(DBConnection.conn(), String.t(), map(), keyword()) :: ResultSet.t()
  def query!(conn, statement, params \\ %{}, opts \\ []) do
    case query(conn, statement, params, opts) do
      {:ok, rs} -> rs
      {:error, err} -> raise err
    end
  end

  @doc """
  Executes a nGQL statement and applies `fun` to every `%Record{}` row,
  returning a list of results.

  Equivalent to:

      {:ok, rs} = query(conn, stmt, params, opts)
      Enum.map(ResultSet.rows(rs), fun)

  ## Example

      NebulaGraphEx.Graph.map(MyApp.Graph, "MATCH (v:Player) RETURN v.name", fn record ->
        record |> NebulaGraphEx.Record.get!("v.name")
      end)
      #=> ["Tim Duncan", "LeBron James", ...]
  """
  @spec map(DBConnection.conn(), String.t(), map(), keyword(), (ResultSet.t() -> term())) ::
          {:ok, [term()]} | {:error, Error.t()}
  def map(conn, statement, params \\ %{}, opts \\ [], fun) do
    case query(conn, statement, params, opts) do
      {:ok, rs} -> {:ok, ResultSet.map(rs, fun)}
      {:error, _} = err -> err
    end
  end

  @doc """
  Checks whether a pool is alive and able to execute a simple query.

  This is an active health check. It first resolves the pool process and then,
  unless disabled, executes `RETURN 1 AS status` to verify the checked-out
  connection can talk to NebulaGraph successfully.

  ## Options

  * `:probe_statement` — query used for the health check. Default: `"RETURN 1 AS status"`.
  * `:probe_query` — whether to run the health-check query. Default: `true`.
  * `:query_opts` — options passed to the probe query. Default: `[]`.

  ## Return value

  On success:

      {:ok,
       %{
         pid: #PID<0.123.0>,
         connected?: true,
         queryable?: true,
         probe_statement: "RETURN 1 AS status"
       }}

  If the pool process is missing:

      {:error,
       %{
         pid: nil,
         connected?: false,
         queryable?: false,
         error: :not_running,
         error_details: %{reason: :not_running, message: "pool process is not running"}
       }}

  If the pool exists but the probe query fails:

      {:error,
       %{
         pid: #PID<0.123.0>,
         connected?: true,
         queryable?: false,
         probe_statement: "RETURN 1 AS status",
         error: %NebulaGraphEx.Error{},
         error_details: %{code: :e_syntax_error, message: "...", category: :query}
       }}
  """
  @spec status(DBConnection.conn(), keyword()) ::
          {:ok,
           %{
             pid: pid(),
             connected?: true,
             queryable?: boolean(),
             probe_statement: String.t() | nil
           }}
          | {:error,
             %{
               pid: pid() | nil,
               connected?: boolean(),
               queryable?: false,
               probe_statement: String.t() | nil,
               error: term(),
               error_details: map()
             }}
  def status(conn, opts \\ []) do
    pid = resolve_pid(conn)

    cond do
      is_nil(pid) ->
        {:error,
         %{
           pid: nil,
           connected?: false,
           queryable?: false,
           probe_statement: nil,
           error: :not_running,
           error_details: error_details(:not_running)
         }}

      not Keyword.get(opts, :probe_query, true) ->
        {:ok, %{pid: pid, connected?: true, queryable?: false, probe_statement: nil}}

      true ->
        probe_statement = Keyword.get(opts, :probe_statement, "RETURN 1 AS status")
        query_opts = Keyword.get(opts, :query_opts, [])

        case query(conn, probe_statement, %{}, query_opts) do
          {:ok, _rs} ->
            {:ok,
             %{
               pid: pid,
               connected?: true,
               queryable?: true,
               probe_statement: probe_statement
             }}

          {:error, error} ->
            {:error,
             %{
               pid: pid,
               connected?: true,
               queryable?: false,
               probe_statement: probe_statement,
               error: error,
               error_details: error_details(error)
             }}
        end
    end
  end

  @doc """
  Issues `USE <space>` on the given connection, switching the active graph
  space for that connection's current session.

  This is only useful when you hold a checked-out connection directly. With
  the pool, prefer passing the `:space` option to `query/4` instead.
  """
  @spec use_space(DBConnection.conn(), String.t(), keyword()) ::
          {:ok, ResultSet.t()} | {:error, Error.t()}
  def use_space(conn, space, opts \\ []) when is_binary(space) do
    query(conn, "USE #{space}", %{}, opts)
  end

  # ─── Private ───────────────────────────────────────────────────────────────

  defp build_db_opts(opts) do
    Keyword.take(opts, [:timeout, :queue_target, :queue_interval, :pool])
  end

  defp error_details(%Error{} = error) do
    %{
      code: error.code,
      code_int: error.code_int,
      message: error.message,
      category: error.category,
      statement: error.statement
    }
  end

  defp error_details(:not_running) do
    %{reason: :not_running, message: "pool process is not running"}
  end

  defp error_details(error) do
    %{reason: error, message: inspect(error)}
  end

  defp resolve_pid(conn) when is_pid(conn), do: conn
  defp resolve_pid(conn), do: GenServer.whereis(conn)

  # ─── Query logging ─────────────────────────────────────────────────────────

  @doc false
  def logging_enabled?(opts) do
    Keyword.get(opts, :log, false) or not is_nil(Keyword.get(opts, :slow_query_threshold))
  end

  defp maybe_log(statement, params, outcome, queue_us, query_us, opts) do
    log = Keyword.get(opts, :log, false)
    threshold_ms = Keyword.get(opts, :slow_query_threshold)
    total_ms = (queue_us + query_us) / 1000

    should_log =
      log or (is_integer(threshold_ms) and total_ms >= threshold_ms) or
        (is_float(threshold_ms) and total_ms >= threshold_ms)

    if should_log do
      level = Keyword.get(opts, :log_level, :debug)
      caller = format_caller(Keyword.get(opts, :nebula_caller))
      do_log(level, statement, params, outcome, queue_us, query_us, caller)
    end
  end

  defp do_log(level, statement, params, outcome, queue_us, query_us, caller) do
    {status, rows} =
      case outcome do
        {:ok, rs} -> {"OK", "#{rs.num_rows} rows"}
        {:error, %Error{code: code}} -> {"ERROR (#{code})", nil}
        _ -> {"ERROR", nil}
      end

    queue_ms = Float.round(queue_us / 1000, 1)
    db_ms = Float.round(query_us / 1000, 1)

    timing =
      if rows,
        do: "#{rows} db=#{db_ms}ms queue=#{queue_ms}ms",
        else: "db=#{db_ms}ms queue=#{queue_ms}ms"

    params_str = if map_size(params) > 0, do: " #{inspect(params)}", else: " []"

    lines =
      [
        if(caller, do: "↳ #{caller}", else: nil),
        "QUERY #{status} #{timing}",
        "#{statement}#{params_str}"
      ]
      |> Enum.reject(&is_nil/1)
      |> Enum.join("\n")

    Logger.log(level, lines)
  end

  defp format_caller(nil), do: nil

  defp format_caller({{:current_stacktrace, frames}, _pool_module}) do
    # Walk past Elixir stdlib, DBConnection, and this library's own frames to
    # find the first frame that belongs to user code.
    skip_prefixes = [
      "Elixir.NebulaGraphEx.",
      "Elixir.DBConnection.",
      "Elixir.Process.",
      ":erlang",
      ":proc_lib",
      ":gen_server",
      ":gen",
      "Elixir.Task."
    ]

    user_frame =
      Enum.find(frames, fn
        {mod, _fun, _arity, _loc} ->
          mod_str = Atom.to_string(mod)
          not Enum.any?(skip_prefixes, &String.starts_with?(mod_str, &1))

        _ ->
          false
      end)

    case user_frame do
      {mod, fun, arity, loc} ->
        file = Keyword.get(loc, :file, "unknown")
        line = Keyword.get(loc, :line, 0)
        "#{inspect(mod)}.#{fun}/#{arity}, at: #{file}:#{line}"

      _ ->
        nil
    end
  end
end