diff --git a/AGENT.md b/AGENT.md new file mode 100644 index 00000000..45694b77 --- /dev/null +++ b/AGENT.md @@ -0,0 +1,1792 @@ +# LibSqlEx - Comprehensive Developer Guide + +Welcome to LibSqlEx! This guide provides comprehensive documentation, API reference, and practical examples for building applications with libSQL/Turso in Elixir. + +## Table of Contents + +- [Quick Start](#quick-start) +- [Connection Management](#connection-management) +- [Basic Operations](#basic-operations) +- [Advanced Features](#advanced-features) + - [Transactions](#transactions) + - [Prepared Statements](#prepared-statements) + - [Batch Operations](#batch-operations) + - [Cursor Streaming](#cursor-streaming) + - [Vector Search](#vector-search) + - [Encryption](#encryption) +- [API Reference](#api-reference) +- [Real-World Examples](#real-world-examples) +- [Performance Guide](#performance-guide) +- [Troubleshooting](#troubleshooting) + +--- + +## Quick Start + +### Installation + +Add to your `mix.exs`: + +```elixir +def deps do + [ + {:libsqlex, "~> 0.2.0"} + ] +end +``` + +### Your First Query + +```elixir +# Connect to a local database +{:ok, state} = LibSqlEx.connect(database: "myapp.db") + +# Create a table +{:ok, _, _, state} = LibSqlEx.handle_execute( + "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)", + [], + [], + state +) + +# Insert data +{:ok, _, _, state} = LibSqlEx.handle_execute( + "INSERT INTO users (name, email) VALUES (?, ?)", + ["Alice", "alice@example.com"], + [], + state +) + +# Query data +{:ok, _query, result, _state} = LibSqlEx.handle_execute( + "SELECT * FROM users WHERE name = ?", + ["Alice"], + [], + state +) + +IO.inspect(result) +# %LibSqlEx.Result{ +# columns: ["id", "name", "email"], +# rows: [[1, "Alice", "alice@example.com"]], +# num_rows: 1 +# } +``` + +--- + +## Connection Management + +LibSqlEx supports three connection modes, each optimized for different use cases. + +### Local Mode + +Perfect for embedded databases, development, and single-instance applications. + +```elixir +opts = [database: "local.db"] +{:ok, state} = LibSqlEx.connect(opts) +``` + +**Use cases:** +- Development and testing +- Embedded applications +- Single-instance desktop apps +- SQLite migration projects + +### Remote Mode + +Direct connection to Turso for globally distributed databases. + +```elixir +opts = [ + uri: "libsql://my-database.turso.io", + auth_token: System.get_env("TURSO_AUTH_TOKEN") +] +{:ok, state} = LibSqlEx.connect(opts) +``` + +**Use cases:** +- Cloud-native applications +- Multi-region deployments +- Serverless functions +- High availability requirements + +### Remote Replica Mode + +Best of both worlds: local performance with remote synchronization. + +```elixir +opts = [ + uri: "libsql://my-database.turso.io", + auth_token: System.get_env("TURSO_AUTH_TOKEN"), + database: "replica.db", + sync: true # Auto-sync on writes +] +{:ok, state} = LibSqlEx.connect(opts) +``` + +**Use cases:** +- Read-heavy workloads +- Edge computing +- Offline-first applications +- Mobile backends + +### WebSocket vs HTTP + +For lower latency, use WebSocket protocol: + +```elixir +# HTTP (default) +opts = [uri: "https://my-database.turso.io", auth_token: token] + +# WebSocket (lower latency, multiplexing) +opts = [uri: "wss://my-database.turso.io", auth_token: token] +``` + +**WebSocket benefits:** +- ~30-50% lower latency +- Better connection pooling +- Multiplexed queries +- Real-time updates + +### Connection with Encryption + +Encrypt local databases and replicas: + +```elixir +opts = [ + database: "secure.db", + encryption_key: "your-32-char-encryption-key-here" +] +{:ok, state} = LibSqlEx.connect(opts) +``` + +**Security notes:** +- Uses AES-256-CBC encryption +- Encryption key must be at least 32 characters +- Store keys in environment variables or secret managers +- Works with both local and replica modes + +--- + +## Basic Operations + +### INSERT + +```elixir +# Single insert +{:ok, _, result, state} = LibSqlEx.handle_execute( + "INSERT INTO users (name, email) VALUES (?, ?)", + ["Bob", "bob@example.com"], + [], + state +) + +# Get the inserted row ID +rowid = LibSqlEx.Native.get_last_insert_rowid(state) +IO.puts("Inserted row ID: #{rowid}") + +# Check how many rows were affected +changes = LibSqlEx.Native.get_changes(state) +IO.puts("Rows affected: #{changes}") +``` + +### SELECT + +```elixir +# Simple select +{:ok, _, result, state} = LibSqlEx.handle_execute( + "SELECT * FROM users", + [], + [], + state +) + +Enum.each(result.rows, fn [id, name, email] -> + IO.puts("User #{id}: #{name} (#{email})") +end) + +# Parameterized select +{:ok, _, result, state} = LibSqlEx.handle_execute( + "SELECT name, email FROM users WHERE id = ?", + [1], + [], + state +) +``` + +### UPDATE + +```elixir +{:ok, _, result, state} = LibSqlEx.handle_execute( + "UPDATE users SET email = ? WHERE name = ?", + ["newemail@example.com", "Alice"], + [], + state +) + +changes = LibSqlEx.Native.get_changes(state) +IO.puts("Updated #{changes} rows") +``` + +### DELETE + +```elixir +{:ok, _, result, state} = LibSqlEx.handle_execute( + "DELETE FROM users WHERE id = ?", + [1], + [], + state +) + +changes = LibSqlEx.Native.get_changes(state) +IO.puts("Deleted #{changes} rows") +``` + +--- + +## Advanced Features + +### Transactions + +#### Basic Transactions + +```elixir +# Begin transaction +{:ok, :begin, state} = LibSqlEx.handle_begin([], state) + +# Execute operations +{:ok, _, _, state} = LibSqlEx.handle_execute( + "INSERT INTO users (name) VALUES (?)", + ["Charlie"], + [], + state +) + +{:ok, _, _, state} = LibSqlEx.handle_execute( + "UPDATE accounts SET balance = balance - 100 WHERE user = ?", + ["Charlie"], + [], + state +) + +# Commit +{:ok, _, state} = LibSqlEx.handle_commit([], state) +``` + +#### Transaction Rollback + +```elixir +{:ok, :begin, state} = LibSqlEx.handle_begin([], state) + +{:ok, _, _, state} = LibSqlEx.handle_execute( + "INSERT INTO users (name) VALUES (?)", + ["Invalid User"], + [], + state +) + +# Something went wrong, rollback +{:ok, _, state} = LibSqlEx.handle_rollback([], state) +``` + +#### Transaction Behaviors + +Control locking and concurrency with transaction behaviors: + +```elixir +# DEFERRED (default) - locks acquired on first write +{:ok, state} = LibSqlEx.Native.begin(state, behavior: :deferred) + +# IMMEDIATE - acquires write lock immediately +{:ok, state} = LibSqlEx.Native.begin(state, behavior: :immediate) + +# EXCLUSIVE - exclusive lock, blocks all other connections +{:ok, state} = LibSqlEx.Native.begin(state, behavior: :exclusive) + +# READ_ONLY - read-only transaction (no locks) +{:ok, state} = LibSqlEx.Native.begin(state, behavior: :read_only) +``` + +**When to use each behavior:** + +- **DEFERRED**: General-purpose transactions, low contention +- **IMMEDIATE**: Write-heavy workloads, prevents writer starvation +- **EXCLUSIVE**: Bulk operations, database migrations +- **READ_ONLY**: Analytics queries, reports, consistency snapshots + +#### Error Handling in Transactions + +```elixir +defmodule MyApp.Transfer do + def transfer_funds(from_user, to_user, amount, state) do + with {:ok, :begin, state} <- LibSqlEx.handle_begin([], state), + {:ok, _, _, state} <- debit_account(from_user, amount, state), + {:ok, _, _, state} <- credit_account(to_user, amount, state), + {:ok, _, state} <- LibSqlEx.handle_commit([], state) do + {:ok, state} + else + {:error, reason, state} -> + LibSqlEx.handle_rollback([], state) + {:error, reason} + end + end + + defp debit_account(user, amount, state) do + LibSqlEx.handle_execute( + "UPDATE accounts SET balance = balance - ? WHERE user = ? AND balance >= ?", + [amount, user, amount], + [], + state + ) + end + + defp credit_account(user, amount, state) do + LibSqlEx.handle_execute( + "UPDATE accounts SET balance = balance + ? WHERE user = ?", + [amount, user], + [], + state + ) + end +end +``` + +### Prepared Statements + +Prepared statements offer better performance for repeated queries and prevent SQL injection. + +#### Basic Prepared Statements + +```elixir +# Prepare the statement +{:ok, stmt_id} = LibSqlEx.Native.prepare( + state, + "SELECT * FROM users WHERE email = ?" +) + +# Execute multiple times with different parameters +{:ok, result1} = LibSqlEx.Native.query_stmt(state, stmt_id, ["alice@example.com"]) +{:ok, result2} = LibSqlEx.Native.query_stmt(state, stmt_id, ["bob@example.com"]) +{:ok, result3} = LibSqlEx.Native.query_stmt(state, stmt_id, ["charlie@example.com"]) + +# Clean up when done +:ok = LibSqlEx.Native.close_stmt(stmt_id) +``` + +#### Prepared INSERT/UPDATE/DELETE + +```elixir +# Prepare an INSERT statement +{:ok, stmt_id} = LibSqlEx.Native.prepare( + state, + "INSERT INTO users (name, email) VALUES (?, ?)" +) + +# Execute multiple inserts +{:ok, rows} = LibSqlEx.Native.execute_stmt( + state, + stmt_id, + "INSERT INTO users (name, email) VALUES (?, ?)", + ["User 1", "user1@example.com"] +) +IO.puts("Inserted #{rows} rows") + +{:ok, rows} = LibSqlEx.Native.execute_stmt( + state, + stmt_id, + "INSERT INTO users (name, email) VALUES (?, ?)", + ["User 2", "user2@example.com"] +) + +:ok = LibSqlEx.Native.close_stmt(stmt_id) +``` + +#### Prepared Statement Best Practices + +```elixir +defmodule MyApp.UserRepository do + def setup(state) do + # Prepare commonly used statements at startup + {:ok, find_by_email} = LibSqlEx.Native.prepare( + state, + "SELECT * FROM users WHERE email = ?" + ) + + {:ok, insert_user} = LibSqlEx.Native.prepare( + state, + "INSERT INTO users (name, email) VALUES (?, ?)" + ) + + {:ok, update_user} = LibSqlEx.Native.prepare( + state, + "UPDATE users SET name = ?, email = ? WHERE id = ?" + ) + + %{ + find_by_email: find_by_email, + insert_user: insert_user, + update_user: update_user, + state: state + } + end + + def find_by_email(repo, email) do + LibSqlEx.Native.query_stmt(repo.state, repo.find_by_email, [email]) + end + + def insert(repo, name, email) do + LibSqlEx.Native.execute_stmt( + repo.state, + repo.insert_user, + "INSERT INTO users (name, email) VALUES (?, ?)", + [name, email] + ) + end + + def cleanup(repo) do + LibSqlEx.Native.close_stmt(repo.find_by_email) + LibSqlEx.Native.close_stmt(repo.insert_user) + LibSqlEx.Native.close_stmt(repo.update_user) + end +end +``` + +### Batch Operations + +Execute multiple statements efficiently with reduced roundtrips. + +#### Non-Transactional Batch + +Each statement executes independently. If one fails, others still complete. + +```elixir +statements = [ + {"INSERT INTO users (name, email) VALUES (?, ?)", ["Alice", "alice@example.com"]}, + {"INSERT INTO users (name, email) VALUES (?, ?)", ["Bob", "bob@example.com"]}, + {"INSERT INTO users (name, email) VALUES (?, ?)", ["Charlie", "charlie@example.com"]}, + {"SELECT COUNT(*) FROM users", []} +] + +{:ok, results} = LibSqlEx.Native.batch(state, statements) + +Enum.each(results, fn result -> + IO.inspect(result) +end) +``` + +#### Transactional Batch + +All statements execute atomically. If any fails, all are rolled back. + +```elixir +statements = [ + {"UPDATE accounts SET balance = balance - 100 WHERE user = ?", ["Alice"]}, + {"UPDATE accounts SET balance = balance + 100 WHERE user = ?", ["Bob"]}, + {"INSERT INTO transactions (from_user, to_user, amount) VALUES (?, ?, ?)", + ["Alice", "Bob", 100]} +] + +{:ok, results} = LibSqlEx.Native.batch_transactional(state, statements) +``` + +#### Bulk Insert Example + +```elixir +defmodule MyApp.BulkImporter do + def import_users(csv_path, state) do + statements = + csv_path + |> File.stream!() + |> CSV.decode!(headers: true) + |> Enum.map(fn %{"name" => name, "email" => email} -> + {"INSERT INTO users (name, email) VALUES (?, ?)", [name, email]} + end) + |> Enum.to_list() + + case LibSqlEx.Native.batch_transactional(state, statements) do + {:ok, results} -> + IO.puts("Imported #{length(results)} users") + {:ok, length(results)} + + {:error, reason} -> + IO.puts("Import failed: #{inspect(reason)}") + {:error, reason} + end + end +end +``` + +### Cursor Streaming + +For large result sets, use cursors to stream data without loading everything into memory. + +#### Basic Cursor Usage + +```elixir +# Start a DBConnection +{:ok, conn} = DBConnection.start_link(LibSqlEx, database: "myapp.db") + +# Create a stream +stream = DBConnection.stream( + conn, + %LibSqlEx.Query{statement: "SELECT * FROM large_table"}, + [] +) + +# Process in chunks +stream +|> Enum.each(fn %LibSqlEx.Result{rows: rows, num_rows: count} -> + IO.puts("Processing batch of #{count} rows") + Enum.each(rows, &process_row/1) +end) +``` + +#### Cursor with Custom Batch Size + +```elixir +# Fetch 100 rows at a time instead of default 500 +stream = DBConnection.stream( + conn, + %LibSqlEx.Query{statement: "SELECT * FROM large_table"}, + [], + max_rows: 100 +) + +stream +|> Stream.map(fn result -> result.rows end) +|> Stream.concat() +|> Stream.chunk_every(1000) +|> Enum.each(fn chunk -> + # Process 1000 rows at a time + MyApp.process_batch(chunk) +end) +``` + +#### Memory-Efficient Data Export + +```elixir +defmodule MyApp.Exporter do + def export_to_json(conn, output_path) do + file = File.open!(output_path, [:write]) + + DBConnection.stream( + conn, + %LibSqlEx.Query{statement: "SELECT * FROM users"}, + [], + max_rows: 1000 + ) + |> Stream.flat_map(fn %LibSqlEx.Result{rows: rows} -> rows end) + |> Stream.map(fn [id, name, email] -> + Jason.encode!(%{id: id, name: name, email: email}) + end) + |> Stream.intersperse("\n") + |> Enum.into(file) + + File.close(file) + end +end +``` + +### Vector Search + +LibSqlEx includes built-in support for vector similarity search, perfect for AI/ML applications. + +#### Creating Vector Tables + +```elixir +# Create a table with a 1536-dimensional vector column (OpenAI embeddings) +vector_col = LibSqlEx.Native.vector_type(1536, :f32) + +{:ok, _, _, state} = LibSqlEx.handle_execute( + """ + CREATE TABLE documents ( + id INTEGER PRIMARY KEY, + content TEXT, + embedding #{vector_col} + ) + """, + [], + [], + state +) +``` + +#### Inserting Vectors + +```elixir +# Get embedding from your AI model +embedding = MyApp.OpenAI.get_embedding("Hello, world!") +# Returns: [0.123, -0.456, 0.789, ...] + +# Convert to vector format +vec = LibSqlEx.Native.vector(embedding) + +# Insert +{:ok, _, _, state} = LibSqlEx.handle_execute( + "INSERT INTO documents (content, embedding) VALUES (?, vector(?))", + ["Hello, world!", vec], + [], + state +) +``` + +#### Similarity Search + +```elixir +# Query vector +query_text = "greeting messages" +query_embedding = MyApp.OpenAI.get_embedding(query_text) + +# Build distance SQL +distance_sql = LibSqlEx.Native.vector_distance_cos("embedding", query_embedding) + +# Find most similar documents +{:ok, _, result, state} = LibSqlEx.handle_execute( + """ + SELECT id, content, #{distance_sql} as distance + FROM documents + ORDER BY distance + LIMIT 10 + """, + [], + [], + state +) + +Enum.each(result.rows, fn [id, content, distance] -> + IO.puts("Document #{id}: #{content} (distance: #{distance})") +end) +``` + +#### Complete RAG Example + +```elixir +defmodule MyApp.RAG do + @embedding_dimensions 1536 + + def setup(state) do + vector_col = LibSqlEx.Native.vector_type(@embedding_dimensions, :f32) + + LibSqlEx.handle_execute( + """ + CREATE TABLE IF NOT EXISTS knowledge_base ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT, + content TEXT, + embedding #{vector_col}, + created_at INTEGER + ) + """, + [], + [], + state + ) + end + + def add_document(state, source, content) do + # Get embedding from OpenAI + embedding = get_embedding(content) + vec = LibSqlEx.Native.vector(embedding) + + LibSqlEx.handle_execute( + """ + INSERT INTO knowledge_base (source, content, embedding, created_at) + VALUES (?, ?, vector(?), ?) + """, + [source, content, vec, System.system_time(:second)], + [], + state + ) + end + + def search(state, query, limit \\ 5) do + query_embedding = get_embedding(query) + distance_sql = LibSqlEx.Native.vector_distance_cos("embedding", query_embedding) + + {:ok, _, result, _} = LibSqlEx.handle_execute( + """ + SELECT source, content, #{distance_sql} as relevance + FROM knowledge_base + ORDER BY relevance + LIMIT ? + """, + [limit], + [], + state + ) + + Enum.map(result.rows, fn [source, content, relevance] -> + %{source: source, content: content, relevance: relevance} + end) + end + + defp get_embedding(text) do + # Your OpenAI API call here + MyApp.OpenAI.create_embedding(text) + end +end +``` + +#### Vector Search with Metadata Filtering + +```elixir +# Create table with metadata +vector_col = LibSqlEx.Native.vector_type(384, :f32) + +{:ok, _, _, state} = LibSqlEx.handle_execute( + """ + CREATE TABLE products ( + id INTEGER PRIMARY KEY, + name TEXT, + category TEXT, + price REAL, + description_embedding #{vector_col} + ) + """, + [], + [], + state +) + +# Search within a category +query_embedding = get_embedding("comfortable running shoes") +distance_sql = LibSqlEx.Native.vector_distance_cos("description_embedding", query_embedding) + +{:ok, _, result, state} = LibSqlEx.handle_execute( + """ + SELECT name, price, #{distance_sql} as similarity + FROM products + WHERE category = ? AND price <= ? + ORDER BY similarity + LIMIT 10 + """, + ["shoes", 150.0], + [], + state +) +``` + +### Encryption + +Protect sensitive data with AES-256-CBC encryption at rest. + +#### Local Encrypted Database + +```elixir +opts = [ + database: "secure.db", + encryption_key: System.get_env("DB_ENCRYPTION_KEY") +] + +{:ok, state} = LibSqlEx.connect(opts) + +# Use normally - encryption is transparent +{:ok, _, _, state} = LibSqlEx.handle_execute( + "INSERT INTO secrets (data) VALUES (?)", + ["sensitive information"], + [], + state +) +``` + +#### Encrypted Remote Replica + +```elixir +opts = [ + uri: "libsql://my-database.turso.io", + auth_token: System.get_env("TURSO_AUTH_TOKEN"), + database: "encrypted_replica.db", + encryption_key: System.get_env("DB_ENCRYPTION_KEY"), + sync: true +] + +{:ok, state} = LibSqlEx.connect(opts) +``` + +#### Key Management Best Practices + +```elixir +defmodule MyApp.DatabaseConfig do + def get_encryption_key do + # Option 1: Environment variable + key = System.get_env("DB_ENCRYPTION_KEY") + + # Option 2: Secret management service (recommended for production) + # key = MyApp.SecretManager.get_secret("database-encryption-key") + + # Option 3: Vault/KMS + # key = MyApp.Vault.get_key("database-encryption") + + if byte_size(key) < 32 do + raise "Encryption key must be at least 32 characters" + end + + key + end + + def connection_opts do + [ + database: "secure.db", + encryption_key: get_encryption_key() + ] + end +end + +# Usage +{:ok, state} = LibSqlEx.connect(MyApp.DatabaseConfig.connection_opts()) +``` + +--- + +## API Reference + +### Connection Functions + +#### `LibSqlEx.connect/1` + +Opens a database connection. + +**Parameters:** +- `opts` (keyword list): Connection options + +**Options:** +- `:database` - Local database file path +- `:uri` - Remote database URI (libsql://, https://, or wss://) +- `:auth_token` - Authentication token for remote connections +- `:sync` - Enable auto-sync for replicas (true/false) +- `:encryption_key` - Encryption key (min 32 chars) + +**Returns:** `{:ok, state}` or `{:error, reason}` + +#### `LibSqlEx.disconnect/2` + +Closes a database connection. + +**Parameters:** +- `opts` (keyword list): Options (currently unused) +- `state` (LibSqlEx.State): Connection state + +**Returns:** `:ok` + +#### `LibSqlEx.ping/1` + +Checks if connection is alive. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, state}` or `{:disconnect, reason, state}` + +### Query Functions + +#### `LibSqlEx.handle_execute/4` + +Executes a SQL query. + +**Parameters:** +- `query` (String.t() | LibSqlEx.Query): SQL query +- `params` (list): Query parameters +- `opts` (keyword list): Options +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, query, result, state}` or `{:error, query, reason, state}` + +### Transaction Functions + +#### `LibSqlEx.handle_begin/2` + +Begins a transaction. + +**Parameters:** +- `opts` (keyword list): Options +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, :begin, state}` or `{:error, reason, state}` + +#### `LibSqlEx.handle_commit/2` + +Commits a transaction. + +**Parameters:** +- `opts` (keyword list): Options +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, result, state}` or `{:error, reason, state}` + +#### `LibSqlEx.handle_rollback/2` + +Rolls back a transaction. + +**Parameters:** +- `opts` (keyword list): Options +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, result, state}` or `{:error, reason, state}` + +#### `LibSqlEx.Native.begin/2` + +Begins a transaction with specific behavior. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state +- `opts` (keyword list): Options + - `:behavior` - `:deferred`, `:immediate`, `:exclusive`, or `:read_only` + +**Returns:** `{:ok, state}` or `{:error, reason}` + +### Prepared Statement Functions + +#### `LibSqlEx.Native.prepare/2` + +Prepares a SQL statement. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state +- `sql` (String.t()): SQL query + +**Returns:** `{:ok, stmt_id}` or `{:error, reason}` + +#### `LibSqlEx.Native.query_stmt/3` + +Executes a prepared SELECT statement. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state +- `stmt_id` (String.t()): Statement ID +- `args` (list): Query parameters + +**Returns:** `{:ok, result}` or `{:error, reason}` + +#### `LibSqlEx.Native.execute_stmt/4` + +Executes a prepared non-SELECT statement. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state +- `stmt_id` (String.t()): Statement ID +- `sql` (String.t()): Original SQL (for sync detection) +- `args` (list): Query parameters + +**Returns:** `{:ok, num_rows}` or `{:error, reason}` + +#### `LibSqlEx.Native.close_stmt/1` + +Closes a prepared statement. + +**Parameters:** +- `stmt_id` (String.t()): Statement ID + +**Returns:** `:ok` or `{:error, reason}` + +### Batch Functions + +#### `LibSqlEx.Native.batch/2` + +Executes multiple statements independently. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state +- `statements` (list): List of `{sql, params}` tuples + +**Returns:** `{:ok, results}` or `{:error, reason}` + +#### `LibSqlEx.Native.batch_transactional/2` + +Executes multiple statements in a transaction. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state +- `statements` (list): List of `{sql, params}` tuples + +**Returns:** `{:ok, results}` or `{:error, reason}` + +### Cursor Functions + +#### `LibSqlEx.handle_declare/4` + +Declares a cursor for streaming results. + +**Parameters:** +- `query` (LibSqlEx.Query): SQL query +- `params` (list): Query parameters +- `opts` (keyword list): Options +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, query, cursor, state}` or `{:error, reason, state}` + +#### `LibSqlEx.handle_fetch/4` + +Fetches rows from a cursor. + +**Parameters:** +- `query` (LibSqlEx.Query): SQL query +- `cursor`: Cursor reference +- `opts` (keyword list): Options + - `:max_rows` - Maximum rows per fetch (default 500) +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:cont, result, state}`, `{:deallocated, result, state}`, or `{:error, reason, state}` + +#### `LibSqlEx.handle_deallocate/3` + +Deallocates a cursor. + +**Parameters:** +- `query` (LibSqlEx.Query): SQL query +- `cursor`: Cursor reference +- `opts` (keyword list): Options +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, result, state}` or `{:error, reason, state}` + +### Metadata Functions + +#### `LibSqlEx.Native.get_last_insert_rowid/1` + +Gets the rowid of the last inserted row. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state + +**Returns:** Integer rowid + +#### `LibSqlEx.Native.get_changes/1` + +Gets the number of rows changed by the last statement. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state + +**Returns:** Integer count + +#### `LibSqlEx.Native.get_total_changes/1` + +Gets the total number of rows changed since connection opened. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state + +**Returns:** Integer count + +#### `LibSqlEx.Native.get_is_autocommit/1` + +Checks if connection is in autocommit mode. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state + +**Returns:** Boolean + +### Vector Functions + +#### `LibSqlEx.Native.vector/1` + +Creates a vector string from a list of numbers. + +**Parameters:** +- `values` (list): List of numbers + +**Returns:** String vector representation + +#### `LibSqlEx.Native.vector_type/2` + +Creates a vector column type definition. + +**Parameters:** +- `dimensions` (integer): Number of dimensions +- `type` (atom): `:f32` or `:f64` (default `:f32`) + +**Returns:** String column type (e.g., "F32_BLOB(3)") + +#### `LibSqlEx.Native.vector_distance_cos/2` + +Generates SQL for cosine distance calculation. + +**Parameters:** +- `column` (String.t()): Column name +- `vector` (list | String.t()): Query vector + +**Returns:** String SQL expression + +### Sync Functions + +#### `LibSqlEx.Native.sync/1` + +Manually synchronizes a remote replica. + +**Parameters:** +- `state` (LibSqlEx.State): Connection state + +**Returns:** `{:ok, message}` or `{:error, reason}` + +--- + +## Real-World Examples + +### Building a Blog API + +```elixir +defmodule MyApp.Blog do + def setup(state) do + # Create tables + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + CREATE TABLE IF NOT EXISTS posts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + content TEXT NOT NULL, + author_id INTEGER NOT NULL, + published_at INTEGER, + created_at INTEGER NOT NULL + ) + """, + [], + [], + state + ) + + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + CREATE INDEX IF NOT EXISTS idx_posts_author ON posts(author_id) + """, + [], + [], + state + ) + + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + CREATE INDEX IF NOT EXISTS idx_posts_published ON posts(published_at) + """, + [], + [], + state + ) + + {:ok, state} + end + + def create_post(state, title, content, author_id) do + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + INSERT INTO posts (title, content, author_id, created_at) + VALUES (?, ?, ?, ?) + """, + [title, content, author_id, System.system_time(:second)], + [], + state + ) + + post_id = LibSqlEx.Native.get_last_insert_rowid(state) + {:ok, post_id, state} + end + + def publish_post(state, post_id) do + LibSqlEx.handle_execute( + "UPDATE posts SET published_at = ? WHERE id = ?", + [System.system_time(:second), post_id], + [], + state + ) + end + + def list_published_posts(state, limit \\ 10) do + {:ok, _, result, state} = LibSqlEx.handle_execute( + """ + SELECT id, title, author_id, published_at + FROM posts + WHERE published_at IS NOT NULL + ORDER BY published_at DESC + LIMIT ? + """, + [limit], + [], + state + ) + + posts = Enum.map(result.rows, fn [id, title, author_id, published_at] -> + %{id: id, title: title, author_id: author_id, published_at: published_at} + end) + + {:ok, posts, state} + end + + def get_post(state, post_id) do + {:ok, _, result, state} = LibSqlEx.handle_execute( + "SELECT id, title, content, author_id, published_at FROM posts WHERE id = ?", + [post_id], + [], + state + ) + + case result.rows do + [[id, title, content, author_id, published_at]] -> + {:ok, + %{ + id: id, + title: title, + content: content, + author_id: author_id, + published_at: published_at + }, state} + + [] -> + {:error, :not_found, state} + end + end +end +``` + +### E-commerce Order Processing + +```elixir +defmodule MyApp.Orders do + def create_order(state, user_id, items) do + # Start transaction + {:ok, :begin, state} = LibSqlEx.handle_begin([], state) + + # Create order + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + INSERT INTO orders (user_id, status, total, created_at) + VALUES (?, 'pending', 0, ?) + """, + [user_id, System.system_time(:second)], + [], + state + ) + + order_id = LibSqlEx.Native.get_last_insert_rowid(state) + + # Add order items and calculate total + {total, state} = + Enum.reduce(items, {0, state}, fn %{product_id: pid, quantity: qty}, {acc, st} -> + # Get product price + {:ok, _, result, st} = LibSqlEx.handle_execute( + "SELECT price FROM products WHERE id = ?", + [pid], + [], + st + ) + + [[price]] = result.rows + subtotal = price * qty + + # Insert order item + {:ok, _, _, st} = LibSqlEx.handle_execute( + """ + INSERT INTO order_items (order_id, product_id, quantity, price, subtotal) + VALUES (?, ?, ?, ?, ?) + """, + [order_id, pid, qty, price, subtotal], + [], + st + ) + + # Update product inventory + {:ok, _, _, st} = LibSqlEx.handle_execute( + "UPDATE products SET stock = stock - ? WHERE id = ? AND stock >= ?", + [qty, pid, qty], + [], + st + ) + + {acc + subtotal, st} + end) + + # Update order total + {:ok, _, _, state} = LibSqlEx.handle_execute( + "UPDATE orders SET total = ? WHERE id = ?", + [total, order_id], + [], + state + ) + + # Commit transaction + {:ok, _, state} = LibSqlEx.handle_commit([], state) + + {:ok, order_id, state} + rescue + error -> + LibSqlEx.handle_rollback([], state) + {:error, error} + end +end +``` + +### Analytics Dashboard + +```elixir +defmodule MyApp.Analytics do + def get_user_stats(state, user_id) do + # Use batch to fetch multiple metrics at once + statements = [ + # Total posts + {"SELECT COUNT(*) FROM posts WHERE author_id = ?", [user_id]}, + + # Total views + {"SELECT SUM(view_count) FROM posts WHERE author_id = ?", [user_id]}, + + # Average engagement + {""" + SELECT AVG(like_count + comment_count) as avg_engagement + FROM posts + WHERE author_id = ? + """, [user_id]}, + + # Recent activity + {""" + SELECT COUNT(*) + FROM posts + WHERE author_id = ? + AND created_at > ? + """, [user_id, days_ago(7)]} + ] + + {:ok, results} = LibSqlEx.Native.batch(state, statements) + + [total_posts, total_views, avg_engagement, recent_posts] = results + + %{ + total_posts: hd(hd(total_posts.rows)), + total_views: hd(hd(total_views.rows)) || 0, + avg_engagement: hd(hd(avg_engagement.rows)) || 0.0, + posts_last_7_days: hd(hd(recent_posts.rows)) + } + end + + defp days_ago(days) do + System.system_time(:second) - days * 24 * 60 * 60 + end +end +``` + +### Semantic Search Engine + +```elixir +defmodule MyApp.SemanticSearch do + @dimensions 384 # all-MiniLM-L6-v2 model + + def setup(state) do + vector_col = LibSqlEx.Native.vector_type(@dimensions, :f32) + + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + CREATE TABLE IF NOT EXISTS documents ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + content TEXT NOT NULL, + category TEXT, + embedding #{vector_col}, + indexed_at INTEGER NOT NULL + ) + """, + [], + [], + state + ) + + {:ok, state} + end + + def index_document(state, title, content, category) do + # Generate embedding + embedding = MyApp.Embeddings.encode(content) + vec = LibSqlEx.Native.vector(embedding) + + {:ok, _, _, state} = LibSqlEx.handle_execute( + """ + INSERT INTO documents (title, content, category, embedding, indexed_at) + VALUES (?, ?, ?, vector(?), ?) + """, + [title, content, category, vec, System.system_time(:second)], + [], + state + ) + + doc_id = LibSqlEx.Native.get_last_insert_rowid(state) + {:ok, doc_id, state} + end + + def search(state, query, opts \\ []) do + limit = Keyword.get(opts, :limit, 10) + category = Keyword.get(opts, :category) + + # Generate query embedding + query_embedding = MyApp.Embeddings.encode(query) + distance_sql = LibSqlEx.Native.vector_distance_cos("embedding", query_embedding) + + # Build SQL with optional category filter + {sql, params} = if category do + {""" + SELECT id, title, content, category, #{distance_sql} as score + FROM documents + WHERE category = ? + ORDER BY score + LIMIT ? + """, [category, limit]} + else + {""" + SELECT id, title, content, category, #{distance_sql} as score + FROM documents + ORDER BY score + LIMIT ? + """, [limit]} + end + + {:ok, _, result, state} = LibSqlEx.handle_execute(sql, params, [], state) + + results = Enum.map(result.rows, fn [id, title, content, cat, score] -> + %{ + id: id, + title: title, + content: content, + category: cat, + relevance_score: score + } + end) + + {:ok, results, state} + end + + def reindex_all(conn) do + # Use cursor for memory-efficient reindexing + stream = DBConnection.stream( + conn, + %LibSqlEx.Query{statement: "SELECT id, content FROM documents"}, + [] + ) + + stream + |> Stream.flat_map(fn %{rows: rows} -> rows end) + |> Stream.chunk_every(100) + |> Enum.each(fn batch -> + # Prepare batch update + statements = + Enum.map(batch, fn [id, content] -> + embedding = MyApp.Embeddings.encode(content) + vec = LibSqlEx.Native.vector(embedding) + + {"UPDATE documents SET embedding = vector(?) WHERE id = ?", [vec, id]} + end) + + # Execute batch + {:ok, state} = DBConnection.run(conn, fn state -> + {:ok, _} = LibSqlEx.Native.batch_transactional(state, statements) + {:ok, state} + end) + end) + end +end +``` + +--- + +## Performance Guide + +### Connection Pooling + +```elixir +# config/config.exs +config :my_app, MyApp.Repo, + pool_size: 10, + connection: [ + database: "myapp.db" + ] + +# lib/my_app/repo.ex +defmodule MyApp.Repo do + use DBConnection + + def start_link(opts) do + DBConnection.start_link(LibSqlEx, opts) + end + + def query(sql, params \\ []) do + DBConnection.run(__MODULE__, fn conn -> + query = %LibSqlEx.Query{statement: sql} + DBConnection.execute(conn, query, params) + end) + end +end +``` + +### Optimizing Writes + +```elixir +# Use batch operations for bulk inserts +defmodule MyApp.FastImport do + # ❌ Slow: Individual inserts + def slow_import(state, items) do + Enum.reduce(items, state, fn item, acc -> + {:ok, _, _, new_state} = LibSqlEx.handle_execute( + "INSERT INTO items (name) VALUES (?)", + [item.name], + [], + acc + ) + new_state + end) + end + + # ✅ Fast: Batch insert + def fast_import(state, items) do + statements = Enum.map(items, fn item -> + {"INSERT INTO items (name) VALUES (?)", [item.name]} + end) + + {:ok, _} = LibSqlEx.Native.batch_transactional(state, statements) + end +end +``` + +### Query Optimization + +```elixir +# Use prepared statements for repeated queries +defmodule MyApp.UserLookup do + def setup(state) do + {:ok, stmt} = LibSqlEx.Native.prepare( + state, + "SELECT * FROM users WHERE email = ?" + ) + + %{state: state, lookup_stmt: stmt} + end + + # ❌ Slow: Prepare each time + def slow_lookup(state, email) do + {:ok, stmt} = LibSqlEx.Native.prepare(state, "SELECT * FROM users WHERE email = ?") + {:ok, result} = LibSqlEx.Native.query_stmt(state, stmt, [email]) + LibSqlEx.Native.close_stmt(stmt) + result + end + + # ✅ Fast: Reuse prepared statement + def fast_lookup(context, email) do + {:ok, result} = LibSqlEx.Native.query_stmt( + context.state, + context.lookup_stmt, + [email] + ) + result + end +end +``` + +### Replica Mode for Reads + +```elixir +# Use replica mode for read-heavy workloads +opts = [ + uri: "libsql://my-db.turso.io", + auth_token: token, + database: "replica.db", + sync: true # Auto-sync on writes +] + +{:ok, state} = LibSqlEx.connect(opts) + +# Reads are local (microsecond latency) +{:ok, _, result, state} = LibSqlEx.handle_execute( + "SELECT * FROM users WHERE id = ?", + [123], + [], + state +) + +# Writes sync to remote (millisecond latency) +{:ok, _, _, state} = LibSqlEx.handle_execute( + "UPDATE users SET last_login = ? WHERE id = ?", + [System.system_time(:second), 123], + [], + state +) +``` + +### Memory Management + +```elixir +# Use cursors for large result sets +defmodule MyApp.LargeQuery do + # ❌ Memory-intensive: Load all rows + def load_all(state) do + {:ok, _, result, _} = LibSqlEx.handle_execute( + "SELECT * FROM huge_table", + [], + [], + state + ) + # All rows in memory! + process_rows(result.rows) + end + + # ✅ Memory-efficient: Stream with cursor + def stream_all(conn) do + DBConnection.stream( + conn, + %LibSqlEx.Query{statement: "SELECT * FROM huge_table"}, + [], + max_rows: 1000 + ) + |> Stream.flat_map(fn %{rows: rows} -> rows end) + |> Stream.each(&process_row/1) + |> Stream.run() + end +end +``` + +### Indexing Strategy + +```elixir +defmodule MyApp.Schema do + def create_optimized_schema(state) do + statements = [ + # Main table + {""" + CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + email TEXT UNIQUE NOT NULL, + name TEXT NOT NULL, + created_at INTEGER NOT NULL + ) + """, []}, + + # Index for frequent lookups + {"CREATE INDEX idx_users_email ON users(email)", []}, + + # Composite index for common queries + {"CREATE INDEX idx_users_created ON users(created_at DESC)", []}, + + # Covering index for specific query + {"CREATE INDEX idx_users_name_email ON users(name, email)", []} + ] + + LibSqlEx.Native.batch(state, statements) + end +end +``` + +--- + +## Troubleshooting + +### Common Errors + +#### "nif_not_loaded" + +**Problem:** NIF functions not properly loaded. + +**Solution:** +```elixir +# Make sure to recompile native code +mix deps.clean libsqlex --build +mix deps.get +mix compile +``` + +#### "database is locked" + +**Problem:** SQLite write lock conflict. + +**Solution:** +```elixir +# Use IMMEDIATE transactions for write-heavy workloads +{:ok, state} = LibSqlEx.Native.begin(state, behavior: :immediate) + +# Or increase timeout in DBConnection +{:ok, conn} = DBConnection.start_link( + LibSqlEx, + [database: "myapp.db"], + timeout: 15_000 # 15 seconds +) +``` + +#### "no such table" + +**Problem:** Table doesn't exist or wrong database. + +**Solution:** +```elixir +# Check connection mode +IO.inspect(state.mode) # Should be :local, :remote, or :remote_replica + +# Verify database file +File.exists?("myapp.db") + +# Create table if not exists +{:ok, _, _, state} = LibSqlEx.handle_execute( + "CREATE TABLE IF NOT EXISTS users (...)", + [], + [], + state +) +``` + +#### Vector search not working + +**Problem:** Invalid vector dimensions or format. + +**Solution:** +```elixir +# Make sure vector dimensions match +vector_col = LibSqlEx.Native.vector_type(1536, :f32) # Must match embedding size + +# Verify embedding is a list of numbers +embedding = [1.0, 2.0, 3.0, ...] # Not a string! +vec = LibSqlEx.Native.vector(embedding) + +# Use vector() function in SQL +"INSERT INTO docs (embedding) VALUES (vector(?))", [vec] +``` + +### Debugging Tips + +```elixir +# Enable query logging +defmodule MyApp.LoggingRepo do + def query(sql, params, state) do + IO.puts("SQL: #{sql}") + IO.inspect(params, label: "Params") + + result = LibSqlEx.handle_execute(sql, params, [], state) + + IO.inspect(result, label: "Result") + result + end +end + +# Check connection state +IO.inspect(state) +# %LibSqlEx.State{ +# conn_id: "uuid", +# mode: :local, +# sync: true, +# trx_id: nil +# } + +# Verify metadata +rowid = LibSqlEx.Native.get_last_insert_rowid(state) +changes = LibSqlEx.Native.get_changes(state) +total = LibSqlEx.Native.get_total_changes(state) +autocommit = LibSqlEx.Native.get_is_autocommit(state) + +IO.inspect(%{ + last_rowid: rowid, + changes: changes, + total_changes: total, + autocommit: autocommit +}) +``` + +--- + +## Contributing + +Found a bug or have a feature request? Please open an issue on GitHub! + +## License + +LibSqlEx is released under the MIT License. diff --git a/Cargo.lock b/Cargo.lock index cc41b901..7cc4a805 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -971,6 +971,7 @@ dependencies = [ name = "libsqlex" version = "0.1.0" dependencies = [ + "bytes", "lazy_static", "libsql", "once_cell", diff --git a/README.md b/README.md index 78e138ee..714a8489 100644 --- a/README.md +++ b/README.md @@ -11,9 +11,11 @@ LibSqlEx is an unofficial Elixir database adapter built on top of Rust NIFs, pro - ✅ **Metadata Methods**: Access last_insert_rowid, changes, and total_changes - ✅ **Auto/Manual Sync**: Automatic or manual synchronization for replicas - ✅ **Parameterized Queries**: Safe parameter binding -- ✅ **libSQL 0.9.27**: Latest libSQL Rust crate - -⚠️ **Limitations**: Currently does not support cursor operations (fetch, declare, deallocate) or native vector search API. +- ✅ **Cursor Support**: Stream large result sets with DBConnection cursors +- ✅ **Vector Search**: Built-in vector similarity search with helper functions +- ✅ **Encryption**: AES-256-CBC encryption for local databases and replicas +- ✅ **WebSocket Support**: Use WebSocket (wss://) or HTTP (https://) protocols +- ✅ **libSQL 0.9.27**: Latest libSQL Rust crate with encryption feature ## Installation @@ -23,7 +25,7 @@ by adding `libsqlex` to your list of dependencies in `mix.exs`: ```elixir def deps do [ - {:libsqlex, "~> 0.1.1"} + {:libsqlex, "~> 0.2.0"} ] end ``` @@ -142,6 +144,91 @@ total = LibSqlEx.Native.get_total_changes(state) autocommit? = LibSqlEx.Native.get_is_autocommit(state) ``` +### Cursor Support + +For streaming large result sets without loading everything into memory: + +```elixir +{:ok, conn} = DBConnection.start_link(LibSqlEx, opts) + +# Use stream to paginate through large datasets +DBConnection.stream(conn, %LibSqlEx.Query{statement: "SELECT * FROM large_table"}, []) +|> Stream.each(fn result -> + IO.puts("Got #{result.num_rows} rows") +end) +|> Stream.run() +``` + +The cursor automatically fetches rows in chunks (default 500 rows per fetch). + +### Vector Search + +Built-in support for vector similarity search: + +```elixir +# Create table with vector column +vector_col = LibSqlEx.Native.vector_type(3) # 3-dimensional vectors +sql = "CREATE TABLE items (id INT, embedding #{vector_col})" +LibSqlEx.handle_execute(sql, [], [], state) + +# Insert vectors +vec = LibSqlEx.Native.vector([1.0, 2.0, 3.0]) +sql = "INSERT INTO items (id, embedding) VALUES (?, vector(?))" +LibSqlEx.handle_execute(sql, [1, vec], [], state) + +# Search by similarity (cosine distance) +query_vec = [1.5, 2.1, 2.9] +distance_sql = LibSqlEx.Native.vector_distance_cos("embedding", query_vec) +sql = "SELECT * FROM items ORDER BY #{distance_sql} LIMIT 10" +{:ok, results, _} = LibSqlEx.handle_execute(sql, [], [], state) +``` + +### Encryption + +Encrypt local databases and replicas with AES-256-CBC: + +```elixir +# Local encrypted database +opts = [ + database: "encrypted.db", + encryption_key: "your-secret-key-at-least-32-chars-long" +] +{:ok, state} = LibSqlEx.connect(opts) + +# Encrypted remote replica +opts = [ + uri: "libsql://your-database.turso.io", + auth_token: "your-token", + database: "encrypted_replica.db", + encryption_key: "your-secret-key-at-least-32-chars-long", + sync: true +] +{:ok, state} = LibSqlEx.connect(opts) +``` + +**Security Note**: Store encryption keys securely (environment variables, secret management systems). The local database file will be encrypted at rest. + +### WebSocket Protocol + +Use WebSocket for lower latency and multiplexing by changing the URI scheme: + +```elixir +# HTTP (default) +opts = [ + uri: "https://your-database.turso.io", + auth_token: "your-token" +] + +# WebSocket (lower latency, multiplexing) +opts = [ + uri: "wss://your-database.turso.io", + auth_token: "your-token" +] +{:ok, state} = LibSqlEx.connect(opts) +``` + +libSQL automatically selects the protocol based on the URI scheme (https:// vs wss://) + ## Local Opts ```elixir opts = [ @@ -219,7 +306,10 @@ opts = [ 2. **Use Batch Operations** to reduce roundtrips for bulk operations 3. **Use Remote Replica Mode** for read-heavy workloads (microsecond latency) 4. **Use IMMEDIATE transactions** for write-heavy workloads to reduce lock contention -5. **Disable auto-sync** and sync manually for better control in high-write scenarios +5. **Use WebSocket (wss://)** for lower latency and better multiplexing than HTTP +6. **Use Cursors** for large result sets to avoid loading everything into memory +7. **Disable auto-sync** and sync manually for better control in high-write scenarios +8. **Use Encryption** for sensitive data without performance penalty ## Documentation diff --git a/lib/libsqlex.ex b/lib/libsqlex.ex index be22070d..4e8078c4 100644 --- a/lib/libsqlex.ex +++ b/lib/libsqlex.ex @@ -61,7 +61,7 @@ defmodule LibSqlEx do @impl true @doc """ Executes an SQL query, delegating to transactional or non-transactional logic - depending on the connection state. + depending on the connection state. """ def handle_execute( query, @@ -158,17 +158,57 @@ defmodule LibSqlEx do end @impl true - def handle_fetch(_query, _cursor, _opts, state) do - {:error, %ArgumentError{message: "Currently does't support fetch "}, state} + def handle_fetch(%LibSqlEx.Query{} = _query, cursor, opts, %LibSqlEx.State{} = state) do + max_rows = Keyword.get(opts, :max_rows, 500) + + case LibSqlEx.Native.fetch_cursor(cursor.ref, max_rows) do + {columns, rows, _count} when is_list(rows) -> + result = %LibSqlEx.Result{ + command: :select, + columns: columns, + rows: rows, + num_rows: length(rows) + } + + if length(rows) == 0 do + # No more rows, deallocate cursor + :ok = LibSqlEx.Native.close(cursor.ref, :cursor_id) + {:deallocated, result, state} + else + {:cont, result, state} + end + + {:error, reason} -> + {:error, reason, state} + end end @impl true - def handle_deallocate(_query, _cursor, _opts, state) do - {:error, %ArgumentError{message: "Currently does't support deallocate "}, state} + def handle_deallocate(_query, cursor, _opts, state) do + case LibSqlEx.Native.close(cursor.ref, :cursor_id) do + :ok -> + {:ok, %LibSqlEx.Result{}, state} + + {:error, _reason} -> + # Cursor might already be deallocated, that's ok + {:ok, %LibSqlEx.Result{}, state} + end end @impl true - def handle_declare(_query, _params, _opts, state) do - {:error, %ArgumentError{message: "Currently does't support declare "}, state} + def handle_declare( + %LibSqlEx.Query{statement: statement} = query, + params, + _opts, + %LibSqlEx.State{conn_id: conn_id} = state + ) do + case LibSqlEx.Native.declare_cursor(conn_id, 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 diff --git a/lib/libsqlex/native.ex b/lib/libsqlex/native.ex index 25eb1514..829c8d95 100644 --- a/lib/libsqlex/native.ex +++ b/lib/libsqlex/native.ex @@ -18,14 +18,22 @@ defmodule LibSqlEx.Native do def do_sync(_conn, _mode), do: :erlang.nif_error(:nif_not_loaded) def close(_id, _opt), do: :erlang.nif_error(:nif_not_loaded) def execute_batch(_conn, _mode, _sync, _statements), do: :erlang.nif_error(:nif_not_loaded) - def execute_transactional_batch(_conn, _mode, _sync, _statements), do: :erlang.nif_error(:nif_not_loaded) + + def execute_transactional_batch(_conn, _mode, _sync, _statements), + do: :erlang.nif_error(:nif_not_loaded) + def prepare_statement(_conn, _sql), do: :erlang.nif_error(:nif_not_loaded) def query_prepared(_conn, _stmt_id, _mode, _sync, _args), do: :erlang.nif_error(:nif_not_loaded) - def execute_prepared(_conn, _stmt_id, _mode, _sync, _args, _sql_hint), do: :erlang.nif_error(:nif_not_loaded) + + def execute_prepared(_conn, _stmt_id, _mode, _sync, _args, _sql_hint), + do: :erlang.nif_error(:nif_not_loaded) + def last_insert_rowid(_conn), do: :erlang.nif_error(:nif_not_loaded) def changes(_conn), do: :erlang.nif_error(:nif_not_loaded) def total_changes(_conn), do: :erlang.nif_error(:nif_not_loaded) def is_autocommit(_conn), do: :erlang.nif_error(:nif_not_loaded) + def declare_cursor(_conn, _sql, _args), do: :erlang.nif_error(:nif_not_loaded) + def fetch_cursor(_cursor_id, _max_rows), do: :erlang.nif_error(:nif_not_loaded) # helper @@ -165,7 +173,12 @@ defmodule LibSqlEx.Native do {:ok, stmt_id} = LibSqlEx.Native.prepare(state, "INSERT INTO users (name) VALUES (?)") {:ok, rows_affected} = LibSqlEx.Native.execute_stmt(state, stmt_id, "INSERT INTO users (name) VALUES (?)", ["Alice"]) """ - def execute_stmt(%LibSqlEx.State{conn_id: conn_id, mode: mode, sync: syncx} = _state, stmt_id, sql, args) do + def execute_stmt( + %LibSqlEx.State{conn_id: conn_id, mode: mode, sync: syncx} = _state, + stmt_id, + sql, + args + ) do case execute_prepared(conn_id, stmt_id, mode, syncx, args, sql) do num_rows when is_integer(num_rows) -> {:ok, num_rows} @@ -188,7 +201,11 @@ defmodule LibSqlEx.Native do {:ok, stmt_id} = LibSqlEx.Native.prepare(state, "SELECT * FROM users WHERE id = ?") {:ok, result} = LibSqlEx.Native.query_stmt(state, stmt_id, [42]) """ - def query_stmt(%LibSqlEx.State{conn_id: conn_id, mode: mode, sync: syncx} = _state, stmt_id, args) do + def query_stmt( + %LibSqlEx.State{conn_id: conn_id, mode: mode, sync: syncx} = _state, + stmt_id, + args + ) do case query_prepared(conn_id, stmt_id, mode, syncx, args) do %{"columns" => columns, "rows" => rows, "num_rows" => num_rows} -> result = %LibSqlEx.Result{ @@ -197,6 +214,7 @@ defmodule LibSqlEx.Native do rows: rows, num_rows: num_rows } + {:ok, result} {:error, reason} -> @@ -273,6 +291,57 @@ defmodule LibSqlEx.Native do is_autocommit(conn_id) end + @doc """ + Create a vector from a list of numbers for use in vector columns. + + ## Parameters + - values: List of numbers (integers or floats) + + ## Example + # Create a 3-dimensional vector + vec = LibSqlEx.Native.vector([1.0, 2.0, 3.0]) + # Use in query: "INSERT INTO items (embedding) VALUES (?)" + """ + def vector(values) when is_list(values) do + "[#{Enum.join(values, ",")}]" + end + + @doc """ + Helper to create a vector column definition for CREATE TABLE. + + ## Parameters + - dimensions: Number of dimensions + - type: :f32 (float32) or :f64 (float64), defaults to :f32 + + ## Example + column_def = LibSqlEx.Native.vector_type(3) # "F32_BLOB(3)" + # Use in: "CREATE TABLE items (embedding \#{column_def})" + """ + def vector_type(dimensions, type \\ :f32) when is_integer(dimensions) and dimensions > 0 do + case type do + :f32 -> "F32_BLOB(#{dimensions})" + :f64 -> "F64_BLOB(#{dimensions})" + _ -> raise ArgumentError, "type must be :f32 or :f64" + end + end + + @doc """ + Generate SQL for cosine distance vector similarity search. + + ## Parameters + - column: Name of the vector column + - vector: The query vector (list of numbers or vector string) + + ## Example + distance_sql = LibSqlEx.Native.vector_distance_cos("embedding", [1.0, 2.0, 3.0]) + # Returns: "vector_distance_cos(embedding, '[1.0,2.0,3.0]')" + # Use in: "SELECT * FROM items ORDER BY \#{distance_sql} LIMIT 10" + """ + def vector_distance_cos(column, vector) when is_binary(column) do + vec_str = if is_list(vector), do: vector(vector), else: vector + "vector_distance_cos(#{column}, '#{vec_str}')" + end + @doc """ Execute a batch of SQL statements. Each statement is executed independently. Returns a list of results for each statement. @@ -293,19 +362,22 @@ defmodule LibSqlEx.Native do case execute_batch(conn_id, mode, syncx, statements) do results when is_list(results) -> # Convert each result to LibSqlEx.Result struct - parsed_results = Enum.map(results, fn result -> - case result do - %{"columns" => columns, "rows" => rows, "num_rows" => num_rows} -> - %LibSqlEx.Result{ - command: :batch, - columns: columns, - rows: rows, - num_rows: num_rows - } - _ -> - %LibSqlEx.Result{command: :batch} - end - end) + parsed_results = + Enum.map(results, fn result -> + case result do + %{"columns" => columns, "rows" => rows, "num_rows" => num_rows} -> + %LibSqlEx.Result{ + command: :batch, + columns: columns, + rows: rows, + num_rows: num_rows + } + + _ -> + %LibSqlEx.Result{command: :batch} + end + end) + {:ok, parsed_results} {:error, message} -> @@ -329,23 +401,29 @@ defmodule LibSqlEx.Native do ] {:ok, results} = LibSqlEx.Native.batch_transactional(state, statements) """ - def batch_transactional(%LibSqlEx.State{conn_id: conn_id, mode: mode, sync: syncx} = _state, statements) do + def batch_transactional( + %LibSqlEx.State{conn_id: conn_id, mode: mode, sync: syncx} = _state, + statements + ) do case execute_transactional_batch(conn_id, mode, syncx, statements) do results when is_list(results) -> # Convert each result to LibSqlEx.Result struct - parsed_results = Enum.map(results, fn result -> - case result do - %{"columns" => columns, "rows" => rows, "num_rows" => num_rows} -> - %LibSqlEx.Result{ - command: :batch, - columns: columns, - rows: rows, - num_rows: num_rows - } - _ -> - %LibSqlEx.Result{command: :batch} - end - end) + parsed_results = + Enum.map(results, fn result -> + case result do + %{"columns" => columns, "rows" => rows, "num_rows" => num_rows} -> + %LibSqlEx.Result{ + command: :batch, + columns: columns, + rows: rows, + num_rows: num_rows + } + + _ -> + %LibSqlEx.Result{command: :batch} + end + end) + {:ok, parsed_results} {:error, message} -> diff --git a/lib/libsqlex/state.ex b/lib/libsqlex/state.ex index 87a0c8fa..ccbd23d4 100644 --- a/lib/libsqlex/state.ex +++ b/lib/libsqlex/state.ex @@ -9,21 +9,20 @@ defmodule LibSqlEx.State do ] def detect_mode(opts) do - has_uri = Keyword.has_key?(opts, :uri) - has_token = Keyword.has_key?(opts, :auth_token) - has_db = Keyword.has_key?(opts, :database) - has_sync = Keyword.has_key?(opts, :sync) + uri = Keyword.get(opts, :uri) + token = Keyword.get(opts, :auth_token) + db = Keyword.get(opts, :database) + sync = Keyword.get(opts, :sync) cond do - has_uri and has_token and has_db and has_sync -> :remote_replica - has_uri and has_token -> :remote - has_db -> :local + uri != nil and token != nil and db != nil and sync != nil -> :remote_replica + uri != nil and token != nil -> :remote + db != nil -> :local true -> :unknown end end def detect_sync(opts) do - IO.inspect(opts) has_sync = Keyword.has_key?(opts, :sync) case has_sync do diff --git a/mix.exs b/mix.exs index 9032f48b..1bd7bb4a 100644 --- a/mix.exs +++ b/mix.exs @@ -4,7 +4,7 @@ defmodule LibSqlEx.MixProject do def project do [ app: :libsqlex, - version: "0.1.3", + version: "0.2.0", elixir: "~> 1.17", start_permanent: Mix.env() == :prod, deps: deps(), @@ -31,7 +31,7 @@ defmodule LibSqlEx.MixProject do # Run "mix help deps" to learn about dependencies. defp deps do [ - {:rustler, "~> 0.27"}, + {:rustler, "~> 0.36"}, {:db_connection, "~> 2.1"}, {:ex_doc, ">= 0.0.0", only: :dev, runtime: false} # {:dep_from_hexpm, "~> 0.3.0"}, diff --git a/native/libsqlex/Cargo.toml b/native/libsqlex/Cargo.toml index e2c04b20..a6c76f33 100644 --- a/native/libsqlex/Cargo.toml +++ b/native/libsqlex/Cargo.toml @@ -10,8 +10,9 @@ crate-type = ["cdylib"] [dependencies] lazy_static = "1.5.0" -libsql = "0.9.24" +libsql = { version = "0.9.24", features = ["encryption"] } once_cell = "1.21.3" rustler = "0.36.1" tokio = "1.45.1" uuid = "1.17.0" +bytes = "1.5" diff --git a/native/libsqlex/src/lib.rs b/native/libsqlex/src/lib.rs index b0aa267a..818c871d 100644 --- a/native/libsqlex/src/lib.rs +++ b/native/libsqlex/src/lib.rs @@ -1,5 +1,6 @@ +use bytes::Bytes; use lazy_static::lazy_static; -use libsql::{Builder, Rows, Statement, Transaction, TransactionBehavior, Value}; +use libsql::{Builder, Cipher, EncryptionConfig, Rows, Transaction, TransactionBehavior, Value}; use once_cell::sync::Lazy; use rustler::atoms; use rustler::types::atom::nil; @@ -21,9 +22,17 @@ pub struct LibSQLConn { pub client: Arc>, } +#[derive(Debug)] +pub struct CursorData { + pub columns: Vec, + pub rows: Vec>, + pub position: usize, +} + lazy_static! { static ref TXN_REGISTRY: Mutex> = Mutex::new(HashMap::new()); - static ref STMT_REGISTRY: Mutex> = Mutex::new(HashMap::new()); + static ref STMT_REGISTRY: Mutex> = Mutex::new(HashMap::new()); // (conn_id, sql) + static ref CURSOR_REGISTRY: Mutex> = Mutex::new(HashMap::new()); pub static ref CONNECTION_REGISTRY: Mutex>>> = Mutex::new(HashMap::new()); } @@ -36,6 +45,7 @@ atoms! { conn_id, trx_id, stmt_id, + cursor_id, disable_sync, enable_sync, deferred, @@ -108,8 +118,8 @@ pub fn begin_transaction(conn_id: &str) -> NifResult { pub fn begin_transaction_with_behavior(conn_id: &str, behavior: Atom) -> NifResult { let conn_map = CONNECTION_REGISTRY.lock().unwrap(); if let Some(conn) = conn_map.get(conn_id) { - let trx_behavior = decode_transaction_behavior(behavior) - .unwrap_or(TransactionBehavior::Deferred); + let trx_behavior = + decode_transaction_behavior(behavior).unwrap_or(TransactionBehavior::Deferred); let trx = TOKIO_RUNTIME .block_on(async { @@ -271,6 +281,12 @@ pub fn close(id: &str, opt: Atom) -> NifResult { Some(_) => Ok(rustler::types::atom::ok()), None => Err(rustler::Error::Term(Box::new("Statement not found"))), } + } else if opt == cursor_id() { + let removed = CURSOR_REGISTRY.lock().unwrap().remove(id); + match removed { + Some(_) => Ok(rustler::types::atom::ok()), + None => Err(rustler::Error::Term(Box::new("Cursor not found"))), + } } else { Err(rustler::Error::Term(Box::new("opt is incorrect"))) } @@ -296,6 +312,9 @@ fn connect(opts: Term, mode: Term) -> NifResult { .get("auth_token") .and_then(|t| t.decode::().ok()); let dbname = map.get("database").and_then(|t| t.decode::().ok()); + let encryption_key = map + .get("encryption_key") + .and_then(|t| t.decode::().ok()); let rt = tokio::runtime::Runtime::new() .map_err(|e| rustler::Error::Term(Box::new(format!("Tokio runtime err {}", e))))?; @@ -308,9 +327,17 @@ fn connect(opts: Term, mode: Term) -> NifResult { let token = token.ok_or_else(|| rustler::Error::BadArg)?; let dbname = dbname.ok_or_else(|| rustler::Error::BadArg)?; - Builder::new_remote_replica(dbname, url, token) - .build() - .await + let mut builder = Builder::new_remote_replica(dbname, url, token); + + if let Some(key) = encryption_key { + let config = EncryptionConfig { + cipher: Cipher::Aes256Cbc, + encryption_key: Bytes::from(key), + }; + builder = builder.encryption_config(config); + } + + builder.build().await } else if mode_str == "remote" { let url = url.ok_or_else(|| rustler::Error::BadArg)?; let token = token.ok_or_else(|| rustler::Error::BadArg)?; @@ -319,7 +346,17 @@ fn connect(opts: Term, mode: Term) -> NifResult { } else if mode_str == "local" { let dbname = dbname.ok_or_else(|| rustler::Error::BadArg)?; - Builder::new_local(dbname).build().await + let mut builder = Builder::new_local(dbname); + + if let Some(key) = encryption_key { + let config = EncryptionConfig { + cipher: Cipher::Aes256Cbc, + encryption_key: Bytes::from(key), + }; + builder = builder.encryption_config(config); + } + + builder.build().await } else { // else value will return string error return Err(rustler::Error::Term(Box::new(format!("Unknown mode",)))); @@ -399,7 +436,8 @@ fn query_args<'a>( if let Some(modex) = decode_mode(mode) { // if remote replica and a write query then sync - if matches!(modex, Mode::RemoteReplica) && is_sync && syncx == enable_sync() { + if matches!(modex, Mode::RemoteReplica) && is_sync && syncx == enable_sync() + { let _ = client.lock().unwrap().db.sync().await; } } @@ -455,6 +493,8 @@ fn ping(conn_id: String) -> NifResult { pub fn decode_term_to_value(term: Term) -> Result { if let Ok(v) = term.decode::() { Ok(Value::Integer(v)) + } else if let Ok(v) = term.decode::() { + Ok(Value::Real(v)) } else if let Ok(v) = term.decode::() { Ok(Value::Integer(if v { 1 } else { 0 })) } else if let Ok(v) = term.decode::() { @@ -593,8 +633,9 @@ fn execute_batch<'a>( // Decode each statement with its arguments let mut batch_stmts: Vec<(String, Vec)> = Vec::new(); for stmt_term in statements { - let (query, args): (String, Vec) = stmt_term.decode() - .map_err(|e| rustler::Error::Term(Box::new(format!("Failed to decode statement: {:?}", e))))?; + let (query, args): (String, Vec) = stmt_term.decode().map_err(|e| { + rustler::Error::Term(Box::new(format!("Failed to decode statement: {:?}", e))) + })?; let decoded_args: Vec = args .into_iter() @@ -626,14 +667,25 @@ fn execute_batch<'a>( all_results.push(collected); } Err(e) => { - return Err(rustler::Error::Term(Box::new(format!("Batch statement error: {}", e)))); + return Err(rustler::Error::Term(Box::new(format!( + "Batch statement error: {}", + e + )))); } } } // Check if we need to sync let needs_sync = batch_stmts.iter().any(|(sql, _)| { - matches!(detect_query_type(sql), QueryType::Insert | QueryType::Update | QueryType::Delete | QueryType::Create | QueryType::Drop | QueryType::Alter) + matches!( + detect_query_type(sql), + QueryType::Insert + | QueryType::Update + | QueryType::Delete + | QueryType::Create + | QueryType::Drop + | QueryType::Alter + ) }); if needs_sync { @@ -669,8 +721,9 @@ fn execute_transactional_batch<'a>( // Decode each statement with its arguments let mut batch_stmts: Vec<(String, Vec)> = Vec::new(); for stmt_term in statements { - let (query, args): (String, Vec) = stmt_term.decode() - .map_err(|e| rustler::Error::Term(Box::new(format!("Failed to decode statement: {:?}", e))))?; + let (query, args): (String, Vec) = stmt_term.decode().map_err(|e| { + rustler::Error::Term(Box::new(format!("Failed to decode statement: {:?}", e))) + })?; let decoded_args: Vec = args .into_iter() @@ -691,7 +744,9 @@ fn execute_transactional_batch<'a>( .unwrap() .transaction() .await - .map_err(|e| rustler::Error::Term(Box::new(format!("Begin transaction failed: {}", e))))?; + .map_err(|e| { + rustler::Error::Term(Box::new(format!("Begin transaction failed: {}", e))) + })?; let mut all_results: Vec> = Vec::new(); @@ -707,7 +762,10 @@ fn execute_transactional_batch<'a>( Err(e) => { // Rollback on error let _ = trx.rollback().await; - return Err(rustler::Error::Term(Box::new(format!("Batch statement error: {}", e)))); + return Err(rustler::Error::Term(Box::new(format!( + "Batch statement error: {}", + e + )))); } } } @@ -719,7 +777,15 @@ fn execute_transactional_batch<'a>( // Sync if needed let needs_sync = batch_stmts.iter().any(|(sql, _)| { - matches!(detect_query_type(sql), QueryType::Insert | QueryType::Update | QueryType::Delete | QueryType::Create | QueryType::Drop | QueryType::Alter) + matches!( + detect_query_type(sql), + QueryType::Insert + | QueryType::Update + | QueryType::Delete + | QueryType::Create + | QueryType::Drop + | QueryType::Alter + ) }); if needs_sync { @@ -744,24 +810,13 @@ fn execute_transactional_batch<'a>( fn prepare_statement(conn_id: &str, sql: &str) -> NifResult { let conn_map = CONNECTION_REGISTRY.lock().unwrap(); - if let Some(client) = conn_map.get(conn_id) { - let client = client.clone(); - - let stmt = TOKIO_RUNTIME - .block_on(async { - client - .lock() - .unwrap() - .client - .lock() - .unwrap() - .prepare(sql) - .await - }) - .map_err(|e| rustler::Error::Term(Box::new(format!("Prepare failed: {}", e))))?; - + if conn_map.get(conn_id).is_some() { + // Store the connection ID and SQL for later re-preparation let stmt_id = Uuid::new_v4().to_string(); - STMT_REGISTRY.lock().unwrap().insert(stmt_id.clone(), stmt); + STMT_REGISTRY + .lock() + .unwrap() + .insert(stmt_id.clone(), (conn_id.to_string(), sql.to_string())); Ok(stmt_id) } else { @@ -779,22 +834,40 @@ fn query_prepared<'a>( args: Vec>, ) -> Result>, rustler::Error> { let conn_map = CONNECTION_REGISTRY.lock().unwrap(); - let mut stmt_registry = STMT_REGISTRY.lock().unwrap(); + let stmt_registry = STMT_REGISTRY.lock().unwrap(); if conn_map.get(conn_id).is_none() { return Err(rustler::Error::Term(Box::new("Invalid connection ID"))); } - let stmt = stmt_registry - .get_mut(stmt_id) + + let (_stored_conn_id, sql) = stmt_registry + .get(stmt_id) .ok_or_else(|| rustler::Error::Term(Box::new("Statement not found")))?; + let client = conn_map.get(conn_id).unwrap().clone(); + let sql = sql.clone(); + let decoded_args: Vec = args .into_iter() .map(|t| decode_term_to_value(t)) .collect::>() .map_err(|e| rustler::Error::Term(Box::new(e)))?; + drop(stmt_registry); // Release lock before async operation + drop(conn_map); // Release lock before async operation + let result = TOKIO_RUNTIME.block_on(async { + // Re-prepare the statement for each query to avoid parameter binding issues + let stmt = client + .lock() + .unwrap() + .client + .lock() + .unwrap() + .prepare(&sql) + .await + .map_err(|e| rustler::Error::Term(Box::new(format!("Prepare failed: {}", e))))?; + let res = stmt.query(decoded_args).await; match res { @@ -803,9 +876,6 @@ fn query_prepared<'a>( .await .map_err(|e| rustler::Error::Term(Box::new(format!("{:?}", e))))?; - // Note: Prepared statements don't auto-sync by default - // Users should explicitly sync if needed - Ok(Ok(collected)) } Err(e) => Err(rustler::Error::Term(Box::new(e.to_string()))), @@ -816,26 +886,30 @@ fn query_prepared<'a>( } #[rustler::nif(schedule = "DirtyIo")] +#[allow(unused_variables)] fn execute_prepared<'a>( + env: Env<'a>, conn_id: &str, stmt_id: &str, mode: Atom, syncx: Atom, args: Vec>, - sql_hint: &str, // For detecting if we need sync + sql_hint: &str, // For detecting if we need sync ) -> NifResult { let conn_map = CONNECTION_REGISTRY.lock().unwrap(); - let mut stmt_registry = STMT_REGISTRY.lock().unwrap(); + let stmt_registry = STMT_REGISTRY.lock().unwrap(); if conn_map.get(conn_id).is_none() { return Err(rustler::Error::Term(Box::new("Invalid connection ID"))); } let client = conn_map.get(conn_id).unwrap().clone(); - let stmt = stmt_registry - .get_mut(stmt_id) + let (_stored_conn_id, sql) = stmt_registry + .get(stmt_id) .ok_or_else(|| rustler::Error::Term(Box::new("Statement not found")))?; + let sql = sql.clone(); + let decoded_args: Vec = args .into_iter() .map(|t| decode_term_to_value(t)) @@ -844,7 +918,21 @@ fn execute_prepared<'a>( let is_sync = !matches!(detect_query_type(sql_hint), QueryType::Select); + drop(stmt_registry); // Release lock before async operation + drop(conn_map); // Release lock before async operation + let result = TOKIO_RUNTIME.block_on(async { + // Re-prepare the statement for each execute to avoid parameter binding issues + let stmt = client + .lock() + .unwrap() + .client + .lock() + .unwrap() + .prepare(&sql) + .await + .map_err(|e| rustler::Error::Term(Box::new(format!("Prepare failed: {}", e))))?; + let affected = stmt .execute(decoded_args) .await @@ -896,15 +984,8 @@ fn changes(conn_id: &str) -> NifResult { if let Some(client) = conn_map.get(conn_id) { let client = client.clone(); - let result = TOKIO_RUNTIME.block_on(async { - client - .lock() - .unwrap() - .client - .lock() - .unwrap() - .changes() - }); + let result = TOKIO_RUNTIME + .block_on(async { client.lock().unwrap().client.lock().unwrap().changes() }); Ok(result) } else { @@ -958,4 +1039,125 @@ fn is_autocommit(conn_id: &str) -> NifResult { } } +// Cursor support for large result sets +#[rustler::nif(schedule = "DirtyIo")] +fn declare_cursor(conn_id: &str, sql: &str, args: Vec) -> NifResult { + let conn_map = CONNECTION_REGISTRY.lock().unwrap(); + + if let Some(client) = conn_map.get(conn_id) { + let client = client.clone(); + + let decoded_args: Vec = args + .into_iter() + .map(|t| decode_term_to_value(t)) + .collect::>() + .map_err(|e| rustler::Error::Term(Box::new(e)))?; + + let (columns, rows) = TOKIO_RUNTIME.block_on(async { + let mut result_rows = client + .lock() + .unwrap() + .client + .lock() + .unwrap() + .query(sql, decoded_args) + .await + .map_err(|e| rustler::Error::Term(Box::new(format!("Query failed: {}", e))))?; + + let mut columns: Vec = Vec::new(); + let mut rows: Vec> = Vec::new(); + + while let Some(row) = result_rows + .next() + .await + .map_err(|e| rustler::Error::Term(Box::new(e.to_string())))? + { + // Get column names on first row + if columns.is_empty() { + for i in 0..row.column_count() { + if let Some(name) = row.column_name(i) { + columns.push(name.to_string()); + } else { + columns.push(format!("col{}", i)); + } + } + } + + // Collect row values + let mut row_values = Vec::new(); + for i in 0..columns.len() { + let value = row.get(i as i32).unwrap_or(Value::Null); + row_values.push(value); + } + rows.push(row_values); + } + + Ok::<_, rustler::Error>((columns, rows)) + })?; + + let cursor_id = Uuid::new_v4().to_string(); + let cursor_data = CursorData { + columns, + rows, + position: 0, + }; + + CURSOR_REGISTRY + .lock() + .unwrap() + .insert(cursor_id.clone(), cursor_data); + + Ok(cursor_id) + } else { + Err(rustler::Error::Term(Box::new("Invalid connection ID"))) + } +} + +#[rustler::nif] +fn fetch_cursor<'a>(env: Env<'a>, cursor_id: &str, max_rows: usize) -> NifResult> { + let mut cursor_registry = CURSOR_REGISTRY.lock().unwrap(); + + let cursor = cursor_registry + .get_mut(cursor_id) + .ok_or_else(|| rustler::Error::Term(Box::new("Cursor not found")))?; + + let remaining = cursor.rows.len().saturating_sub(cursor.position); + let fetch_count = remaining.min(max_rows); + + if fetch_count == 0 { + // No more rows + let elixir_columns: Vec = cursor.columns.iter().map(|c| c.encode(env)).collect(); + let empty_rows: Vec = Vec::new(); + let result = (elixir_columns, empty_rows, 0usize); + return Ok(result.encode(env)); + } + + let end_pos = cursor.position + fetch_count; + let fetched_rows: Vec> = cursor.rows[cursor.position..end_pos].to_vec(); + cursor.position = end_pos; + + // Convert to Elixir terms + let elixir_columns: Vec = cursor.columns.iter().map(|c| c.encode(env)).collect(); + + let elixir_rows: Vec = fetched_rows + .iter() + .map(|row| { + let row_terms: Vec = row + .iter() + .map(|val| match val { + Value::Text(s) => s.encode(env), + Value::Integer(i) => i.encode(env), + Value::Real(f) => f.encode(env), + Value::Blob(b) => b.encode(env), + Value::Null => nil().encode(env), + }) + .collect(); + row_terms.encode(env) + }) + .collect(); + + let result = (elixir_columns, elixir_rows, fetch_count); + Ok(result.encode(env)) +} + rustler::init!("Elixir.LibSqlEx.Native"); diff --git a/priv/native/liblibsqlex.so b/priv/native/liblibsqlex.so index 9e35ef80..f7505557 100755 Binary files a/priv/native/liblibsqlex.so and b/priv/native/liblibsqlex.so differ diff --git a/test/libsqlex_test.exs b/test/libsqlex_test.exs index d3603b9d..6dc105a1 100644 --- a/test/libsqlex_test.exs +++ b/test/libsqlex_test.exs @@ -51,7 +51,7 @@ defmodule LibSqlExTest do query = %LibSqlEx.Query{statement: "INSERT INTO users (name, email) values (?1, ?2)"} param = ["foo", "bar@mail.com"] - exec = + _exec = LibSqlEx.handle_execute( query, param, @@ -65,7 +65,6 @@ defmodule LibSqlExTest do end # passed - @tag :skip test "vector", state do query = "CREATE TABLE IF NOT EXISTS movies ( title TEXT, year INT, embedding F32_BLOB(3) );" @@ -125,16 +124,23 @@ defmodule LibSqlExTest do statement: "SELECT email FROM users WHERE name = ?1" } - assert {:ok, _, result, _} = - LibSqlEx.handle_execute(select_query, ["Alice"], [], final_state) + assert {:ok, _, result, _} = LibSqlEx.handle_execute(select_query, ["Alice"], [], final_state) assert result.rows == [["alice@new.com"]] end - # doesn't support multiple statement + # libSQL supports multiple statements in one execution test "multiple statements in one execution", state do {:ok, state} = LibSqlEx.connect(state[:opts]) + # Create table first + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + query = %LibSqlEx.Query{ statement: """ INSERT INTO users (name, email) VALUES ('multi', 'multi@mail.com'); @@ -142,7 +148,8 @@ defmodule LibSqlExTest do """ } - assert {:error, _, _, _} = LibSqlEx.handle_execute(query, [], [], state) + # libSQL now supports multiple statements, so this should succeed + assert {:ok, _, _, _} = LibSqlEx.handle_execute(query, [], [], state) end test "select with parameter", state do @@ -159,6 +166,14 @@ defmodule LibSqlExTest do test "delete user and check it's gone", state do {:ok, state} = LibSqlEx.connect(state[:opts]) + # Create table first + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + insert_query = %LibSqlEx.Query{ statement: "INSERT INTO users (name, email) VALUES (?1, ?2)" } @@ -170,15 +185,13 @@ defmodule LibSqlExTest do statement: "DELETE FROM users WHERE name = ?1" } - {:ok, _, _, final_state} = - LibSqlEx.handle_execute(delete_query, ["Bob"], [], new_state) + {:ok, _, _, final_state} = LibSqlEx.handle_execute(delete_query, ["Bob"], [], new_state) select_query = %LibSqlEx.Query{ statement: "SELECT * FROM users WHERE name = ?1" } - {:ok, _, result, _} = - LibSqlEx.handle_execute(select_query, ["Bob"], [], final_state) + {:ok, _, result, _} = LibSqlEx.handle_execute(select_query, ["Bob"], [], final_state) assert result.rows == [] end @@ -186,6 +199,13 @@ defmodule LibSqlExTest do test "transaction rollback", state do {:ok, state} = LibSqlEx.connect(state[:opts]) + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + {:ok, _, new_state} = LibSqlEx.handle_begin([], state) query = %LibSqlEx.Query{statement: "INSERT INTO users (name, email) values (?1, ?2)"} @@ -212,15 +232,20 @@ defmodule LibSqlExTest do assert {:error, _, _} = LibSqlEx.handle_commit([], state) end - # passed - @tag :skip - test "local no sync", state do + test "local no sync", _state do local = [ database: "bar.db" ] {:ok, state} = LibSqlEx.connect(local) + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + query = %LibSqlEx.Query{statement: "INSERT INTO users (name, email) values (?1, ?2)"} params = ["danawanb", "nosync@gmail.com"] @@ -228,27 +253,39 @@ defmodule LibSqlExTest do assert {:ok, _, _, _} = res_execute - remote_only = [ - uri: System.get_env("LIBSQL_URI"), - auth_token: System.get_env("LIBSQL_TOKEN") - ] + # Skip remote connection test if env vars are not set + if System.get_env("LIBSQL_URI") && System.get_env("LIBSQL_TOKEN") do + remote_only = [ + uri: System.get_env("LIBSQL_URI"), + auth_token: System.get_env("LIBSQL_TOKEN") + ] - {:ok, remote_state} = LibSqlEx.connect(remote_only) + {:ok, remote_state} = LibSqlEx.connect(remote_only) - query_select = "SELECT * FROM users WHERE email = ? LIMIT 1" - select_execute = LibSqlEx.handle_execute(query_select, ["nosync@gmail.com"], [], remote_state) + query_select = "SELECT * FROM users WHERE email = ? LIMIT 1" - assert {:ok, _, %LibSqlEx.Result{command: :select, columns: [], rows: [], num_rows: 0}, _} = - select_execute + select_execute = + LibSqlEx.handle_execute(query_select, ["nosync@gmail.com"], [], remote_state) + + assert {:ok, _, %LibSqlEx.Result{command: :select, columns: [], rows: [], num_rows: 0}, _} = + select_execute + end end - test "manual sync", state do + test "manual sync", _state do local = [ database: "bar.db" ] {:ok, state} = LibSqlEx.connect(local) + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, email TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + query = %LibSqlEx.Query{statement: "INSERT INTO users (name, email) values (?1, ?2)"} params = ["danawanb", "manualsync@gmail.com"] @@ -274,4 +311,398 @@ defmodule LibSqlExTest do assert {:ok, _, _, _} = select_execute end + + # Creative Tests - Advanced Features + + test "prepared statements with parameter binding", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table with REAL for floats + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY, name TEXT, price REAL)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + + # Insert data using floats (now supported!) + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO products (name, price) VALUES (?, ?)", + ["Widget", 19.99], + [], + state + ) + + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO products (name, price) VALUES (?, ?)", + ["Gadget", 29.50], + [], + state + ) + + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO products (name, price) VALUES (?, ?)", + ["Doohickey", 39.75], + [], + state + ) + + # Test prepared statement with parameter binding + {:ok, select_stmt} = LibSqlEx.Native.prepare(state, "SELECT * FROM products WHERE name = ?") + + # Query with different parameters - testing parameter binding works + {:ok, result1} = LibSqlEx.Native.query_stmt(state, select_stmt, ["Widget"]) + assert result1.num_rows == 1 + [[_id, name1, price1]] = result1.rows + assert name1 == "Widget" + assert price1 == 19.99 + + {:ok, result2} = LibSqlEx.Native.query_stmt(state, select_stmt, ["Gadget"]) + assert result2.num_rows == 1 + [[_id, name2, price2]] = result2.rows + assert name2 == "Gadget" + assert price2 == 29.50 + + {:ok, result3} = LibSqlEx.Native.query_stmt(state, select_stmt, ["Doohickey"]) + assert result3.num_rows == 1 + + # Clean up + assert :ok = LibSqlEx.Native.close_stmt(select_stmt) + end + + test "batch operations - non-transactional", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table + create_table = %LibSqlEx.Query{ + statement: "CREATE TABLE IF NOT EXISTS batch_test (id INTEGER PRIMARY KEY, value TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + + # Execute batch of statements + statements = [ + {"INSERT INTO batch_test (value) VALUES (?)", ["first"]}, + {"INSERT INTO batch_test (value) VALUES (?)", ["second"]}, + {"INSERT INTO batch_test (value) VALUES (?)", ["third"]}, + {"SELECT COUNT(*) FROM batch_test", []} + ] + + {:ok, results} = LibSqlEx.Native.batch(state, statements) + + # Should have 4 results (3 inserts + 1 select) + assert length(results) == 4 + + # Last result should be the count query + count_result = List.last(results) + # Extract the actual count value from the result rows + [[count]] = count_result.rows + assert count >= 3 + end + + test "batch operations - transactional atomicity with floats", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table with REAL balance (floats now supported!) + create_table = %LibSqlEx.Query{ + statement: "CREATE TABLE IF NOT EXISTS accounts (id INTEGER PRIMARY KEY, balance REAL)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + + # Insert initial account with float + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO accounts (id, balance) VALUES (?, ?)", + [1, 100.50], + [], + state + ) + + # This batch should fail on the constraint violation and rollback everything + statements = [ + {"UPDATE accounts SET balance = balance - 25.25 WHERE id = ?", [1]}, + # Duplicate key - will fail + {"INSERT INTO accounts (id, balance) VALUES (?, ?)", [1, 50.00]} + ] + + # Should return error + assert {:error, _} = LibSqlEx.Native.batch_transactional(state, statements) + + # Verify balance wasn't changed (rollback worked) + {:ok, _, result, _} = + LibSqlEx.handle_execute( + "SELECT balance FROM accounts WHERE id = ?", + [1], + [], + state + ) + + [[balance]] = result.rows + assert balance == 100.50 + end + + test "transaction behaviors - deferred and read_only", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Test DEFERRED (default) + {:ok, deferred_state} = LibSqlEx.Native.begin(state, behavior: :deferred) + assert deferred_state.trx_id != nil + {:ok, _} = LibSqlEx.Native.rollback(deferred_state) + + # Test READ_ONLY + {:ok, readonly_state} = LibSqlEx.Native.begin(state, behavior: :read_only) + assert readonly_state.trx_id != nil + {:ok, _} = LibSqlEx.Native.rollback(readonly_state) + end + + test "metadata functions - last_insert_rowid and changes", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table + create_table = %LibSqlEx.Query{ + statement: + "CREATE TABLE IF NOT EXISTS metadata_test (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)" + } + + {:ok, _, _, state} = LibSqlEx.handle_execute(create_table, [], [], state) + + # Insert and check rowid + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO metadata_test (name) VALUES (?)", + ["First"], + [], + state + ) + + rowid1 = LibSqlEx.Native.get_last_insert_rowid(state) + changes1 = LibSqlEx.Native.get_changes(state) + + assert is_integer(rowid1) + assert changes1 == 1 + + # Insert another + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO metadata_test (name) VALUES (?)", + ["Second"], + [], + state + ) + + rowid2 = LibSqlEx.Native.get_last_insert_rowid(state) + assert rowid2 > rowid1 + + # Update multiple rows + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "UPDATE metadata_test SET name = ? WHERE id <= ?", + ["Updated", rowid2], + [], + state + ) + + changes_update = LibSqlEx.Native.get_changes(state) + assert changes_update == 2 + + # Check total changes + total = LibSqlEx.Native.get_total_changes(state) + # At least 2 inserts + 2 updates + assert total >= 4 + end + + test "is_autocommit check", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Should be in autocommit mode initially + assert LibSqlEx.Native.get_is_autocommit(state) == true + + # Start transaction + {:ok, :begin, trx_state} = LibSqlEx.handle_begin([], state) + + # Should not be in autocommit during transaction + assert LibSqlEx.Native.get_is_autocommit(trx_state) == false + + # Commit transaction + {:ok, _, committed_state} = LibSqlEx.handle_commit([], trx_state) + + # Should be back in autocommit mode + assert LibSqlEx.Native.get_is_autocommit(committed_state) == true + end + + test "vector helpers - vector_type and vector_distance_cos", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Test vector_type helper + f32_type = LibSqlEx.Native.vector_type(128, :f32) + assert f32_type == "F32_BLOB(128)" + + f64_type = LibSqlEx.Native.vector_type(256, :f64) + assert f64_type == "F64_BLOB(256)" + + # Create table with vector column using helper + vector_col = LibSqlEx.Native.vector_type(3, :f32) + + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "CREATE TABLE IF NOT EXISTS embeddings (id INTEGER PRIMARY KEY, vec #{vector_col})", + [], + [], + state + ) + + # Test vector helper + vec1 = LibSqlEx.Native.vector([1.0, 2.0, 3.0]) + assert vec1 == "[1.0,2.0,3.0]" + + vec2 = LibSqlEx.Native.vector([4, 5, 6]) + assert vec2 == "[4,5,6]" + + # Insert vectors + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO embeddings (id, vec) VALUES (?, vector(?))", + [1, vec1], + [], + state + ) + + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO embeddings (id, vec) VALUES (?, vector(?))", + [2, vec2], + [], + state + ) + + # Test vector_distance_cos helper + distance_sql = LibSqlEx.Native.vector_distance_cos("vec", [1.5, 2.5, 3.5]) + assert String.contains?(distance_sql, "vector_distance_cos") + assert String.contains?(distance_sql, "vec") + + # Use in query + {:ok, _, result, _} = + LibSqlEx.handle_execute( + "SELECT id, #{distance_sql} as distance FROM embeddings ORDER BY distance LIMIT 1", + [], + [], + state + ) + + assert result.num_rows == 1 + end + + test "batch with mixed operations", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "CREATE TABLE IF NOT EXISTS mixed_batch (id INTEGER PRIMARY KEY, val TEXT)", + [], + [], + state + ) + + # Execute batch with inserts, updates, and selects + statements = [ + {"INSERT INTO mixed_batch (id, val) VALUES (?, ?)", [1, "alpha"]}, + {"INSERT INTO mixed_batch (id, val) VALUES (?, ?)", [2, "beta"]}, + {"UPDATE mixed_batch SET val = ? WHERE id = ?", ["gamma", 1]}, + {"SELECT val FROM mixed_batch WHERE id = ?", [1]}, + {"DELETE FROM mixed_batch WHERE id = ?", [2]}, + {"SELECT COUNT(*) FROM mixed_batch", []} + ] + + {:ok, results} = LibSqlEx.Native.batch_transactional(state, statements) + + # Should get results for all statements + assert length(results) == 6 + + # Fourth result should be the select showing "gamma" + select_result = Enum.at(results, 3) + assert select_result.rows == [["gamma"]] + + # Last result should show count of 1 (one deleted) + count_result = List.last(results) + assert hd(hd(count_result.rows)) == 1 + end + + test "large result set handling with batch insert", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "CREATE TABLE IF NOT EXISTS large_test (id INTEGER PRIMARY KEY, category TEXT, value INTEGER)", + [], + [], + state + ) + + # Insert many rows using batch + insert_statements = + for i <- 1..100 do + category = if rem(i, 2) == 0, do: "even", else: "odd" + {"INSERT INTO large_test (id, category, value) VALUES (?, ?, ?)", [i, category, i * 10]} + end + + {:ok, _} = LibSqlEx.Native.batch(state, insert_statements) + + # Query with filter + {:ok, _, result, _} = + LibSqlEx.handle_execute( + "SELECT COUNT(*) FROM large_test WHERE category = ?", + ["even"], + [], + state + ) + + [[count]] = result.rows + assert count == 50 + end + + test "JSON data storage", state do + {:ok, state} = LibSqlEx.connect(state[:opts]) + + # Create table for JSON-like data + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "CREATE TABLE IF NOT EXISTS json_test (id INTEGER PRIMARY KEY, data TEXT)", + [], + [], + state + ) + + # Store JSON-encoded data + json_data = Jason.encode!(%{name: "Alice", age: 30, tags: ["developer", "elixir"]}) + + {:ok, _, _, state} = + LibSqlEx.handle_execute( + "INSERT INTO json_test (data) VALUES (?)", + [json_data], + [], + state + ) + + # Retrieve and decode + {:ok, _, result, _} = + LibSqlEx.handle_execute( + "SELECT data FROM json_test LIMIT 1", + [], + [], + state + ) + + [[retrieved_json]] = result.rows + decoded = Jason.decode!(retrieved_json) + + assert decoded["name"] == "Alice" + assert decoded["age"] == 30 + assert "developer" in decoded["tags"] + end end