defmodule EctoLibSql do @moduledoc """ DBConnection implementation for LibSQL and Turso databases. This module provides the core database connection functionality for LibSQL/Turso, implementing the `DBConnection` behaviour. It supports three connection modes: - **Local**: SQLite files on the local filesystem - **Remote**: Direct connections to Turso cloud databases - **Remote Replica**: Local SQLite files that sync with remote Turso databases ## Features - Connection management via Rust NIFs for high performance - Full transaction support with multiple isolation levels - Query execution in both transactional and non-transactional contexts - Cursor support for streaming large result sets - Prepared statement caching - Batch operations (transactional and non-transactional) - Vector similarity search - Database encryption ## Connection Options - `:database` - Path to local SQLite database file - `:uri` - Remote LibSQL server URI (e.g., `"libsql://your-db.turso.io"`) - `:auth_token` - Authentication token for remote connections - `:sync` - Enable automatic sync for embedded replicas (boolean) - `:encryption_key` - Encryption key for local database (min 32 characters) ## Examples # Local database {:ok, conn} = DBConnection.start_link(EctoLibSql, database: "local.db") # Remote Turso database {:ok, conn} = DBConnection.start_link(EctoLibSql, uri: "libsql://your-db.turso.io", auth_token: "your-token" ) # Embedded replica (local + remote sync) {:ok, conn} = DBConnection.start_link(EctoLibSql, database: "local.db", uri: "libsql://your-db.turso.io", auth_token: "your-token", sync: true ) """ use DBConnection # Default busy timeout in milliseconds (5 seconds) @default_busy_timeout 5000 @impl true @doc """ Opens a connection to LibSQL using the native Rust layer. Returns `{:ok, state}` on success or `{:error, reason}` on failure. Automatically uses remote replica if the opts provided database, uri, and auth token. ## Options - `:database` - Path to local SQLite database file - `:uri` - Remote LibSQL server URI (e.g., `"libsql://your-db.turso.io"`) - `:auth_token` - Authentication token for remote connections - `:sync` - Enable automatic sync for embedded replicas (boolean) - `:encryption_key` - Encryption key for local database (min 32 characters) - `:busy_timeout` - Busy timeout in milliseconds (default: 5000) Controls how long SQLite waits for locks before returning SQLITE_BUSY. Set to 0 to disable (not recommended for production). """ @spec connect(Keyword.t()) :: {:ok, EctoLibSql.State.t()} | {:error, term()} def connect(opts) do mode = EctoLibSql.State.detect_mode(opts) case EctoLibSql.Native.connect(opts, mode) do conn_id when is_binary(conn_id) -> state = %EctoLibSql.State{ conn_id: conn_id, mode: mode, sync: EctoLibSql.State.detect_sync(opts) } # Set busy_timeout for better concurrency handling busy_timeout = Keyword.get(opts, :busy_timeout, @default_busy_timeout) case EctoLibSql.Native.set_busy_timeout(conn_id, busy_timeout) do :ok -> {:ok, state} {:error, reason} -> # Log warning but don't fail connection - busy_timeout is an optimisation require Logger Logger.warning("Failed to set busy_timeout: #{inspect(reason)}") {:ok, state} end {:error, _} = err -> err other -> {:error, {:unexpected_response, other}} end end @impl true @doc """ Pings the current connection to ensure it is still alive. """ @spec ping(EctoLibSql.State.t()) :: {:ok, EctoLibSql.State.t()} | {:disconnect, :ping_failed, EctoLibSql.State.t()} def ping(%EctoLibSql.State{conn_id: conn_id} = state) do case EctoLibSql.Native.ping(conn_id) do true -> {:ok, state} _ -> {:disconnect, :ping_failed, state} end end @impl true @doc """ Disconnects from the database by closing the underlying native connection. Removes the connection from the Rust connection registry and cleans up any resources. """ @spec disconnect(term(), EctoLibSql.State.t()) :: :ok | {:error, term(), EctoLibSql.State.t()} def disconnect(_opts, %EctoLibSql.State{conn_id: conn_id} = state) do EctoLibSql.Native.close_conn(conn_id, :conn_id, state) end @impl true @doc """ Executes an SQL query, delegating to transactional or non-transactional logic depending on the connection state. """ @spec handle_execute( EctoLibSql.Query.t() | String.t(), list(), Keyword.t(), EctoLibSql.State.t() ) :: {:ok, EctoLibSql.Query.t(), EctoLibSql.Result.t(), EctoLibSql.State.t()} | {:error, EctoLibSql.Error.t(), EctoLibSql.State.t()} def handle_execute(query, args, _opts, %EctoLibSql.State{trx_id: trx_id} = state) do query_struct = case query do %EctoLibSql.Query{} -> query query when is_binary(query) -> %EctoLibSql.Query{statement: query} end # Check if query returns rows (SELECT, EXPLAIN, WITH, RETURNING clauses). # If so, route through query path instead of execute path. sql = query_struct.statement case EctoLibSql.Native.should_use_query_path(sql) do true -> # Query returns rows, use the query path. # Convert map arguments to list if needed (NIFs expect lists). normalised_args = normalise_args_for_query(sql, args) if trx_id do EctoLibSql.Native.query_with_trx_args(trx_id, state.conn_id, sql, normalised_args) |> format_query_result(state) else EctoLibSql.Native.query_args( state.conn_id, state.mode, state.sync, sql, normalised_args ) |> format_query_result(state) end false -> # Query doesn't return rows, use the execute path (INSERT/UPDATE/DELETE). # Note: execute_with_trx and execute_non_trx handle argument normalisation internally. if trx_id do EctoLibSql.Native.execute_with_trx(state, query_struct, args) else EctoLibSql.Native.execute_non_trx(query_struct, state, args) end end end # Helper to format raw query results for return defp format_query_result(%{"columns" => columns, "rows" => rows, "num_rows" => num_rows}, state) do result = %EctoLibSql.Result{ columns: columns, rows: rows, num_rows: num_rows } {:ok, %EctoLibSql.Query{}, result, state} end defp format_query_result({:error, reason}, state) do error = build_error(reason) {:error, error, state} end # Build an EctoLibSql.Error from various reason formats. defp build_error(%EctoLibSql.Error{} = error), do: error defp build_error(reason) when is_binary(reason) do %EctoLibSql.Error{message: reason, sqlite: %{code: :error, message: reason}} end defp build_error(reason) when is_map(reason) do message = Map.get(reason, :message) || Map.get(reason, "message") || inspect(reason) %EctoLibSql.Error{message: message, sqlite: %{code: :error, message: message}} end defp build_error(reason) do message = inspect(reason) %EctoLibSql.Error{message: message, sqlite: %{code: :error, message: message}} end # Convert map arguments to a list by extracting named parameters from SQL. # If args is already a list, return it unchanged. defp normalise_args_for_query(_sql, args) when is_list(args), do: args defp normalise_args_for_query(sql, args) when is_map(args) do # Extract named parameters from SQL in order of appearance. # Supports :name, $name, and @name formats. param_names = extract_named_params(sql) # Validate that all parameters exist in the map and collect missing ones. missing_params = Enum.filter(param_names, fn name -> not has_param?(args, name) end) # Raise error if any parameters are missing. if missing_params != [] do missing_list = Enum.map_join(missing_params, ", ", &":#{&1}") raise ArgumentError, "Missing required parameters: #{missing_list}. " <> "SQL requires: #{Enum.map_join(param_names, ", ", &":#{&1}")}" end # Convert map values to list in parameter order. Enum.map(param_names, fn name -> get_param_value(args, name) end) end # Check if a parameter exists in the map (supports both atom and string keys). # Uses String.to_existing_atom/1 to avoid atom table exhaustion from dynamic SQL. defp has_param?(map, name) when is_binary(name) do # Try existing atom key first (common case), then string key. try do atom_key = String.to_existing_atom(name) Map.has_key?(map, atom_key) or Map.has_key?(map, name) rescue ArgumentError -> # Atom doesn't exist, check string key only. Map.has_key?(map, name) end end # Get a parameter value from a map, supporting both atom and string keys. # Uses String.to_existing_atom/1 to avoid atom table exhaustion from dynamic SQL. # This assumes the parameter exists (validated by has_param?). defp get_param_value(map, name) when is_binary(name) do # Try existing atom key first (common case), then string key. atom_key = String.to_existing_atom(name) Map.get(map, atom_key, Map.get(map, name)) rescue ArgumentError -> # Atom doesn't exist, try string key only. Map.get(map, name) end # Extract named parameter names from SQL in order of appearance. # Returns a list of strings to avoid atom table exhaustion from dynamic SQL. # # LIMITATION: This regex-based approach cannot distinguish between parameter-like # patterns in SQL string literals or comments and actual parameters. For example: # # SELECT ':not_a_param', name FROM users WHERE id = :actual_param # # Would extract both "not_a_param" and "actual_param", even though the first # appears in a string literal. This is an edge case that would require a full # SQL parser to handle correctly (tracking quoted strings, escaped characters, # and comment blocks). In practice, this limitation rarely causes issues because: # 1. SQL string literals containing parameter-like patterns are uncommon # 2. The validation will catch truly missing parameters # 3. Extra entries in the parameter list are ignored during binding # # If this becomes problematic, consider using a proper SQL parser or the # prepared statement introspection approach used in lib/ecto_libsql/native.ex. defp extract_named_params(sql) do # Match :name, $name, or @name patterns. ~r/[:$@]([a-zA-Z_][a-zA-Z0-9_]*)/ |> Regex.scan(sql) |> Enum.map(fn [_full, name] -> name end) end @impl true @doc """ Begins a new database transaction. The transaction behaviour (deferred/immediate/exclusive) can be controlled via options passed to the Native module. """ def handle_begin(_opts, state) do case EctoLibSql.Native.begin(state) do {:ok, new_state} -> {:ok, :begin, new_state} {:error, reason} -> {:error, reason, state} end end @impl true def handle_commit(_opts, %EctoLibSql.State{trx_id: nil} = state) do {:error, %RuntimeError{message: "no active transaction"}, state} end @impl true @doc """ Commits the current transaction. The state must contain a valid transaction ID. For embedded replicas with auto-sync enabled, this will also trigger a sync to the remote database. """ def handle_commit(_opts, state) do case EctoLibSql.Native.commit( %EctoLibSql.State{conn_id: conn_id, trx_id: _trx_id, mode: mode} = state ) do {:ok, _} -> {:ok, %EctoLibSql.Result{}, %EctoLibSql.State{conn_id: conn_id, mode: mode}} {:error, reason} -> {:disconnect, reason, state} end end @impl true @doc """ Rolls back the current transaction. Discards all changes made within the transaction and returns the connection to autocommit mode. """ def handle_rollback(_opts, %EctoLibSql.State{conn_id: conn_id, trx_id: _trx_id} = state) do case EctoLibSql.Native.rollback(state) do {:ok, _} -> {:ok, %EctoLibSql.Result{}, %EctoLibSql.State{conn_id: conn_id, trx_id: nil}} {:error, reason} -> {:disconnect, reason, state} end end @impl true @doc """ Closes the query. Currently a no-op as queries are stateless. """ def handle_close(_query, _opts, state) do {:ok, %EctoLibSql.Result{}, state} end @impl true @doc """ Checks the current transaction status. """ def handle_status(_opts, %EctoLibSql.State{conn_id: _conn_id, trx_id: trx_id} = state) do case trx_id do nil -> # No active transaction, connection is idle {:idle, state} trx_id when is_binary(trx_id) -> # Check if transaction is still active case EctoLibSql.Native.handle_status_transaction(trx_id) do :ok -> {:transaction, state} {:error, _message} -> {:idle, %{state | trx_id: nil}} end end end @impl true @doc """ Prepares a query for execution. Returns the query unchanged as preparation is handled during execution. """ def handle_prepare(%EctoLibSql.Query{} = query, _opts, state) do {:ok, query, state} end @impl true @doc """ Checks out a connection from the pool by verifying it's still alive. """ def checkout(%EctoLibSql.State{conn_id: conn_id} = state) do case EctoLibSql.Native.ping(conn_id) do true -> {:ok, state} {:error, reason} -> {:disconnect, reason, state} end end @impl true @doc """ Fetches the next batch of rows from a cursor. Cursors are used for streaming large result sets without loading everything into memory at once. Automatically deallocates the cursor when no more rows are available. """ def handle_fetch( %EctoLibSql.Query{} = _query, cursor, opts, %EctoLibSql.State{conn_id: conn_id} = state ) do max_rows = Keyword.get(opts, :max_rows, 500) case EctoLibSql.Native.fetch_cursor(conn_id, cursor.ref, max_rows) do {columns, rows, _count} when is_list(rows) -> result = %EctoLibSql.Result{ command: :select, columns: columns, rows: rows, num_rows: length(rows) } if rows == [] do # No more rows, signal halt {:halt, result, state} else {:cont, result, state} end {:error, reason} -> {:error, reason, state} end end @impl true @doc """ Deallocates a cursor, freeing its resources. """ def handle_deallocate(_query, cursor, _opts, state) do case EctoLibSql.Native.close(cursor.ref, :cursor_id) do :ok -> {:ok, %EctoLibSql.Result{}, state} {:error, _reason} -> # Cursor might already be deallocated, that's ok {:ok, %EctoLibSql.Result{}, state} end end @impl true @doc """ Declares a cursor for streaming query results. Cursors allow you to iterate through large result sets in chunks, which is more memory-efficient than loading all rows at once. """ def handle_declare( %EctoLibSql.Query{statement: statement} = query, params, _opts, %EctoLibSql.State{conn_id: conn_id, trx_id: trx_id} = state ) do # Use transaction ID if in a transaction, otherwise use connection ID id = trx_id || conn_id id_type = if trx_id, do: :transaction, else: :connection case EctoLibSql.Native.declare_cursor_with_context(conn_id, id, id_type, statement, params) do cursor_id when is_binary(cursor_id) -> cursor = %{ref: cursor_id} {:ok, query, cursor, state} {:error, reason} -> {:error, reason, state} end end end