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 ` 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 ` 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