bolty

# SPDX-FileCopyrightText: 2025 bolty contributors
# SPDX-License-Identifier: Apache-2.0

defmodule Bolty do
  @moduledoc """
  Bolt driver for Elixir — talk to Neo4j (and other Bolt servers) over the
  binary Bolt protocol.

  Bolty is built on [DBConnection](https://hexdocs.pm/db_connection), so a
  connection is a supervised, pooled process: start one with `start_link/1`
  (or via `child_spec/1` in a supervision tree) and query it with `query/4` /
  `query_many/4`.

  ## Example

  ```elixir
  {:ok, conn} =
    Bolty.start_link(
      hostname: "localhost",
      auth: [username: "neo4j", password: "password"]
    )

  {:ok, %Bolty.Response{} = res} =
    Bolty.query(conn, "MATCH (n:Person) RETURN n.name AS name LIMIT $limit", %{limit: 5})

  # A Response is Enumerable over its result rows
  names = Enum.map(res, & &1["name"])
  ```

  See `start_link/1` for the full connection options (TLS, pooling, auth). Every
  failure — connection, TLS, version negotiation, and Neo4j server errors — is
  surfaced as a `Bolty.Error`, so callers can match a single error shape.

  ## Guides

  - [Public API & compatibility boundary](public_api.md) — what is supported
    under semantic versioning
  - [Telemetry](telemetry.md) — the `[:bolty, :query]` span and other events
  - `usage-rules.md` — an agent-facing quick reference (shipped in the package)
  """

  @type conn() :: DBConnection.conn()

  @typedoc """
  The basic authentication scheme relies on traditional username and password

  * `:username` - Username (required)

  * `:password` - Password (default: `nil`)
  """
  @type basic_auth() ::
          {:username, String.t()}
          | {:password, String.t() | nil}

  @type start_option() ::
          {:uri, String.t()}
          | {:hostname, String.t()}
          | {:port, :inet.port_number()}
          | {:scheme, String.t()}
          | {:routing, boolean()}
          | {:versions, list(String.t() | float())}
          | {:auth, basic_auth()}
          | {:user_agent, String.t()}
          | {:notifications_minimum_severity, String.t()}
          | {:notifications_disabled_categories, list(String.t())}
          | {:ssl_opts, [:ssl.tls_client_option()]}
          | {:connect_timeout, timeout()}
          | {:recv_timeout, timeout()}
          | {:socket_options, [:gen_tcp.connect_option()]}
          | DBConnection.start_option()

  @type option :: %{
          bookmarks: list(),
          mode: String.t(),
          db: String.t() | nil,
          tx_metadata: map() | nil
        }

  @doc """
  Starts the connection process and connects to a Bolt/Neo4j server.

  ## Options

  * `:uri` - Connection URI, of the form `<SCHEME>://<HOST>[:<PORT>[?policy=<POLICY-NAME>]]`.
   Explicit `:hostname`, `:port` and `:scheme` options take priority over the
   corresponding URI components.

  * `:hostname` - Server hostname (default: `"localhost"`)

  * `:port` - Server port (default: `7687`)

  * `:scheme` - Is one among neo4j, neo4j+s, neo4j+ssc, bolt, bolt+s, bolt+ssc
   (default: bolt+s). TLS is derived entirely from this: `bolt`/`neo4j` is
   plaintext, `+s` is full certificate verification against the OS trust
   store, `+ssc` is encrypted but trust-all (self-signed certs). The scheme
   also sets the default for `:routing` (below).

  * `:routing` - Whether to enable server-side routing (SSR) against a Neo4j
   Enterprise causal cluster, so the server forwards each statement to the
   right member based on the query's `:mode`/`:bookmarks` (see `query/4`).
   Defaults to `true` for `neo4j*` schemes and `false` for `bolt*` schemes,
   matching their meaning across the ecosystem; set it explicitly to override
   (`routing: false` for a `neo4j://` URI used cosmetically against a single
   instance or proxy, `routing: true` to force it on a `bolt://` scheme). See
   the [Clustering guide](clustering.html) for prerequisites and caveats.

  * `:versions` - List of Bolt versions to offer during negotiation, as strings
   (`["5.4", "6.0"]`) or `{major, minor..minor}` range tuples (`[{5, 6..8}]`).
   Floats (`[5.4]`) are **deprecated** — they can't distinguish `5.10` from `5.1`
   — and are still accepted with a one-time warning. (Default: the latest
   supported versions.)

  * `:auth` - The basic authentication scheme

  * `:user_agent` - Optionally override the default user agent name. (Default: 'bolty/<version>')

  * `:notifications_minimum_severity` - Set the minimum severity for notifications the server
   should send to the client. Disabling severities allows the server to skip analysis for those,
  which can speed up query execution. (default: nil) _New in neo4j v5.7 and Bolt v5.2_

  * `:notifications_disabled_categories` - Set categories of notifications the server should not
   send to the client. Disabling categories allows the server to skip analysis for those, which
  can speed up query execution. (default: nil) _New in neo4j v5.7 and Bolt v5.2_

  * `:connect_timeout` - Socket connect timeout in milliseconds (default:
      `15_000`)

  * `:recv_timeout` - Per-read timeout in milliseconds for post-connect protocol
      reads (default: `15_000`). This is a socket-inactivity timeout: it fires
      only if the server sends *no* bytes for this long, and the window resets on
      each chunk received, so streaming a large result is unaffected. A read that
      times out is surfaced as `{:error, %Bolty.Error{code: :timeout}}` and the
      connection is disconnected. Can be overridden per query via the same option
      to `query/4`; set to `:infinity` for queries that may compute for a long
      time before returning any data.

  * `:ssl_opts` - A list of `:ssl.connect/2` options, merged *over* the strict,
   secure-by-default TLS options bolty derives from `:scheme` (default: `[]`).
   An explicit `verify:`/`cacertfile:`/`cacerts:` here always overrides the
   scheme-derived default. Only applies when `:scheme` enables TLS (`+s` or
   `+ssc`); ignored for plain `bolt`/`neo4j`.

  The given options are passed down to DBConnection, some of the most commonly used ones are
   documented below:

  * `:after_connect` - A function to run after the connection has been established, either a
      1-arity fun, a `{module, function, args}` tuple, or `nil` (default: `nil`)

  * `:pool` - The pool module to use, defaults to built-in pool provided by DBconnection

  * `:pool_size` - The size of the pool
  """
  @spec start_link([start_option()]) :: GenServer.on_start()
  def start_link(options) do
    DBConnection.start_link(Bolty.Connection, options)
  end

  @doc """
  Returns a supervisor child specification for a DBConnection pool.
  """
  @spec child_spec([start_option()]) :: :supervisor.child_spec()
  def child_spec(options) do
    DBConnection.child_spec(Bolty.Connection, options)
  end

  @doc """
  Executes a single query and returns the result.

  Returns `{:ok, %Bolty.Response{}}` on success. On failure it returns
  `{:error, %Bolty.Error{}}` — every driver-side failure (connection, TLS,
  version negotiation, and Neo4j server errors) is surfaced as a `Bolty.Error`,
  so callers can match a single error shape. Pool checkout/timeout failures may
  additionally surface as `DBConnection.ConnectionError`.

  ## Options

  * `:db` - Target database for multi-database routing (default: server default).

  * `:mode` - Access mode for this query, `"w"` (write, default) or `"r"`
      (read). Under server-side routing (a `neo4j://` connection to a cluster —
      see `start_link/1`'s `:routing` and the [Clustering guide](clustering.html))
      the server uses this to forward the query to a writable or a read member.
      Without routing it is informational only.

  * `:bookmarks` - A list of bookmark strings for causal consistency: the server
      will not run this query until it has caught up to the transactions those
      bookmarks identify. Each `Bolty.Response` carries the bookmark of the
      transaction that produced it; thread the previous response's bookmark into
      the next query to chain reads-after-writes. There is no session
      abstraction, so bookmarks must be passed manually.

  * `:recv_timeout` - Overrides the connection's `:recv_timeout` for this query
      (see `start_link/1`). Pass `:infinity` for a query expected to run longer
      than the connection default before returning data.

  ## Examples

  ```elixir
  {:ok, result} = Bolty.query(conn, "MATCH (n) RETURN n LIMIT 1")

  {:ok, people} = Bolty.query(conn, "MATCH (n:PERSON) RETURN n", %{}, [db: "mydb"])
  ```
  """
  @spec query(conn(), String.t(), map(), keyword()) ::
          {:ok, Bolty.Response.t()} | {:error, Bolty.Error.t() | DBConnection.ConnectionError.t()}
  def query(conn, statement, params \\ %{}, opts \\ []) do
    formatted_params = Map.new(params)

    extra =
      opts
      |> Keyword.take([:bookmarks, :mode, :db, :tx_metadata])
      # Convert to map
      |> Enum.into(%{})

    query = %Bolty.Query{statement: statement, extra: extra}

    do_query(conn, query, formatted_params, opts)
  end

  @doc """
  Executes a single query and returns the result.

  ## Examples

  ```elixir
  result = Bolty.query!(conn, "MATCH (n) RETURN n LIMIT 1")

  people = Bolty.query!(conn, "MATCH (n:PERSON) RETURN n", %{}, [db: "mydb"])
  ```
  """
  @spec query!(conn(), String.t(), map(), keyword()) :: Bolty.Response.t()
  def query!(conn, statement, params \\ %{}, opts \\ []) do
    case query(conn, statement, params, opts) do
      {:ok, result} -> result
      {:error, exception} -> raise exception
    end
  end

  @doc """
  Executes a batch of `;`-separated statements and returns a list of results.

  The batch string is split into individual statements before each is run in
  turn against the same connection. Execution stops at the first failure and
  returns that `{:error, %Bolty.Error{}}`.

  Statements are split on a top-level `;` — one that is not inside a string
  literal, a comment (`//` or `/* */`), or a backtick-quoted identifier — so a
  `;` hiding inside any of those is safe. The terminating `;` is dropped (Bolt
  runs one bare statement at a time), and blank or comment-only segments are
  skipped rather than sent to the server.
  """
  @spec query_many(conn(), String.t(), map(), keyword()) ::
          {:ok, [Bolty.Response.t()]}
          | {:error, Bolty.Error.t() | DBConnection.ConnectionError.t()}
  def query_many(conn, statement, params \\ %{}, opts \\ []) do
    formatted_params = Map.new(params)

    queries = %Bolty.Queries{statement: statement}
    do_query(conn, queries, formatted_params, opts)
  end

  @doc """
  Same as `query_many/4`, but returns the list of `Bolty.Response` structs
  directly and raises the `Bolty.Error` from the first failing statement.
  """
  @spec query_many!(conn(), String.t(), map(), keyword()) :: [Bolty.Response.t()]
  def query_many!(conn, statement, params \\ %{}, opts \\ []) do
    case query_many(conn, statement, params, opts) do
      {:ok, result} -> result
      {:error, exception} -> raise exception
    end
  end

  @doc """
  Lazily streams a query's result in batches, with server-side backpressure.

  Returns a `DBConnection` stream that pulls records from the server in batches
  as it is enumerated, rather than materialising the whole result up front like
  `query/4`. Each element is a `%Bolty.Response{}` for one batch — its `results`
  and `records` are that batch's records; the summary metadata (`stats`,
  `bookmark`, …) lands on the final batch. Enumerating only part of the stream
  fetches only the batches consumed; the rest is discarded server-side.

  **Must be called inside a `transaction/4`** — streaming pages against the
  statement's server-assigned query id, which only exists within an explicit
  transaction. Enumerate the stream within the same transaction callback.

  Unlike `query/4`, a failure mid-stream is surfaced by **raising** a
  `Bolty.Error` while the stream is being enumerated (the `Enumerable` protocol
  has no error-tuple channel), rather than returning `{:error, _}`. The error
  value is the same `%Bolty.Error{}`; the connection is RESET-recovered and
  returns clean to the pool, and the enclosing transaction rolls back.

  ## Options

  * `:fetch_size` - Records to pull per batch (the Bolt `PULL n`). Default
      `1000`. Smaller batches mean more round-trips but lower peak memory.

  * `:mode`, `:bookmarks`, `:db` - As documented on `query/4`, applied to the
      streamed statement's `RUN`.

  ## Example

  ```elixir
  Bolty.transaction(conn, fn conn ->
    conn
    |> Bolty.stream("MATCH (n:Person) RETURN n", %{}, fetch_size: 500)
    |> Stream.flat_map(& &1.results)
    |> Enum.each(&process/1)
  end)
  ```
  """
  @spec stream(DBConnection.t(), String.t(), map(), keyword()) :: DBConnection.Stream.t()
  def stream(conn, statement, params \\ %{}, opts \\ []) do
    formatted_params = Map.new(params)

    extra =
      opts
      |> Keyword.take([:bookmarks, :mode, :db, :tx_metadata])
      |> Enum.into(%{})

    query = %Bolty.Query{statement: statement, extra: extra}

    DBConnection.stream(conn, query, formatted_params, opts)
  end

  @doc """
  Runs `fun` inside a Bolt transaction, committing on return and rolling back on
  `rollback/2` or a raised error.

  `opts` are passed to `DBConnection.transaction/3`. `extra_parameters` is a map
  applied to the opening `BEGIN` for the whole transaction:

  * `:mode` - `"w"` (write, default) or `"r"` (read). Under server-side routing
    (a `neo4j://` connection to a cluster — see `start_link/1`'s `:routing` and
    the [Clustering guide](clustering.html)) the server routes the transaction to
    a writable or a read member accordingly.

  * `:bookmarks` - A list of bookmark strings for causal consistency; the
    transaction will not begin until the server has caught up to the transactions
    they identify. Read a completed transaction's bookmark from the last
    `Bolty.Response` it produced and thread it into the next one — there is no
    session abstraction, so this is manual.

  * `:db` - Target database for the transaction (default: server default).

  ## Example

  ```elixir
  Bolty.transaction(conn, fn conn ->
    Bolty.query!(conn, "CREATE (:Person {name: $n})", %{n: "Ada"})
  end, [], %{mode: "w"})
  ```
  """
  @spec transaction(conn, (DBConnection.t() -> result), [DBConnection.option()], map()) ::
          {:ok, result} | {:error, any}
        when result: var
  def transaction(conn, fun, opts \\ [], extra_parameters \\ %{}) do
    DBConnection.transaction(conn, fun, opts ++ [extra_parameters: extra_parameters])
  end

  @spec rollback(DBConnection.t(), any()) :: no_return()
  defdelegate rollback(conn, reason), to: DBConnection

  @doc """
  Returns metadata about the negotiated connection.

  Can be called with the `conn` passed into a `transaction/3` callback or any
  checked-out connection.

  ## Example

  ```elixir
  Bolty.transaction(Bolt, fn conn ->
    Bolty.connection_info(conn)
    # => %{bolt_version: "5.8", server_version: "Neo4j/5.26.27", policy: %Bolty.Policy{...}}
  end)
  ```
  """
  @spec connection_info(conn()) :: %{
          bolt_version: String.t(),
          server_version: String.t(),
          policy: Bolty.Policy.t()
        }
  def connection_info(conn) do
    case DBConnection.prepare_execute(conn, %Bolty.ConnectionInfo{}, %{}) do
      {:ok, _query, info} -> info
      {:error, error} -> raise error
    end
  end

  # Wrap the round-trip in a `[:bolty, :query]` telemetry span so operators get
  # `:start`/`:stop`/`:exception` events with duration, statement, and outcome
  # without instrumenting every call site. See `guides/telemetry.md`.
  #
  # This span is for the EAGER path only: `:telemetry.span` closes when the fun
  # returns, and here that means the full result is materialised. Lazy streaming
  # (#59, via handle_declare/handle_fetch) must NOT reuse this — a cursor returns
  # before any rows are pulled, so the span would time only RUN and would stuff
  # the whole materialised result into `:result`. Streaming needs its own
  # start-on-declare / stop-on-deallocate instrumentation carrying row counts.
  defp do_query(conn, query, params, options) do
    metadata = %{
      db_system: "neo4j",
      db_statement: statement(query),
      db_instance: options[:db]
    }

    # `:telemetry.span` fires `:exception` (and reraises) if the fun raises/exits;
    # for the `{:error, _}` return we flag a coarse `:result` status and stash the
    # error so tracers can set span status without us holding the whole result.
    :telemetry.span([:bolty, :query], metadata, fn ->
      case DBConnection.prepare_execute(conn, query, params, options) do
        {:ok, _query, result} ->
          {{:ok, result}, Map.put(metadata, :result, :ok)}

        # prepare_execute always surfaces failures as an exception struct
        # (%Bolty.Error{} or %DBConnection.ConnectionError{}).
        {:error, error} = failure ->
          {failure, metadata |> Map.put(:result, :error) |> Map.put(:error, error)}
      end
    end)
  end

  defp statement(%Bolty.Query{statement: statement}), do: statement
  defp statement(%Bolty.Queries{statement: statement}), do: statement
end