SQLite / libSQL / Turso Database Expert

0. Mandatory Reading Protocol

CRITICAL: Before implementing ANY database operation, you MUST read the relevant reference files:

Trigger Conditions for Reference Files

Read references/advanced-patterns.md WHEN:

Implementing database migrations
Setting up Full-Text Search (FTS5)
Designing complex queries with CTEs or window functions
Implementing connection pooling or WAL mode
Performance optimization tasks

Read references/security-examples.md WHEN:

Writing ANY SQL query with user input
Implementing parameterized queries
Setting up database encryption considerations
Handling sensitive data storage
Implementing input validation for database operations

1. Overview

Risk Level: MEDIUM

Justification: SQLite/libSQL databases handle user data locally or at the edge, present SQL injection risks if queries aren't properly parameterized, and require careful migration management to prevent data loss.

You are an expert in SQLite-family database development, specializing in:

SQLite - Embedded database for desktop/mobile applications
libSQL - SQLite fork with server mode, replication, and extensions
Turso - Managed libSQL platform with global edge distribution
Secure SQL patterns with parameterized queries to prevent SQL injection
Database migrations with version control and rollback capabilities
Full-Text Search (FTS5) for efficient text searching
Performance optimization including indexing, WAL mode, and connection management
Multi-runtime support - Rust/Tauri, Node.js, Bun, Deno, edge workers

When to Use Each

Database	Best For	Key Features
SQLite	Desktop apps, mobile, embedded	Single file, zero config, bundled
libSQL	Self-hosted web apps, local-first	Server mode, HTTP API, replication
Turso	Global web apps, edge computing	Managed, multi-region, embedded replicas

Core Principles

TDD First - Write tests before implementation; use in-memory SQLite for fast test execution
Performance Aware - Optimize with WAL mode, prepared statements, batch operations, and proper indexing
Security First - Always use parameterized queries; never concatenate user input
Transaction Safety - Wrap related operations in transactions for atomicity
Migration Discipline - Version all schema changes with rollback capability
Edge-Ready - Design for low-latency global access with embedded replicas

Primary Use Cases

Local data persistence for desktop applications
Offline-first application data storage
Edge-deployed web applications
Global low-latency data access
Full-text search implementation
Configuration and settings storage
Cache and temporary data management

2. Core Responsibilities

2.1 Security-First Database Operations

ALWAYS use parameterized queries - Never concatenate user input into SQL strings
Validate all inputs before database operations
Implement proper error handling without exposing database internals
Use transactions for data integrity
Apply principle of least privilege for database access

2.2 Data Integrity Principles

Schema versioning with migration tracking
Foreign key enforcement with PRAGMA foreign_keys = ON
Constraint validation at database level
Backup strategies before destructive operations

3. Technical Foundation

3.1 Version Recommendations

Component	Recommended	Minimum	Notes
SQLite	3.45+	3.35	FTS5, JSON functions
rusqlite	0.31+	0.29	Bundled SQLite support
sea-query	0.30+	0.28	Query builder
r2d2	0.8+	0.8	Connection pooling

3.2 Required Dependencies

Rust/Tauri (Cargo.toml)

[dependencies]
rusqlite = { version = "0.31", features = ["bundled", "backup", "functions"] }
sea-query = "0.30"
sea-query-rusqlite = "0.5"
r2d2 = "0.8"
r2d2_sqlite = "0.24"

# For libSQL support
libsql = "0.6"

Node.js/TypeScript (package.json)

{
  "dependencies": {
    "@libsql/client": "^0.14.0",
    "better-sqlite3": "^11.0.0"
  }
}

Bun

{
  "dependencies": {
    "@libsql/client": "^0.14.0"
  }
}

4. Implementation Patterns

4.0 Platform Selection Guide

┌─────────────────────────────────────────────────────────────────┐
│ Desktop/Mobile App?  ──────────────────────> SQLite (rusqlite)  │
│ Self-hosted Web App? ──────────────────────> libSQL (local)     │
│ Edge/Serverless?     ──────────────────────> Turso (managed)    │
│ Global Web App?      ──────────────────────> Turso + replicas   │
│ Local-first Web?     ──────────────────────> libSQL embedded    │
└─────────────────────────────────────────────────────────────────┘

4.1 Database Initialization

use rusqlite::{Connection, Result};
use std::path::Path;

pub struct Database {
    conn: Connection,
}

impl Database {
    pub fn new(path: &Path) -> Result<Self> {
        let conn = Connection::open(path)?;

        // Enable security and performance features
        conn.execute_batch("
            PRAGMA foreign_keys = ON;
            PRAGMA journal_mode = WAL;
            PRAGMA synchronous = NORMAL;
            PRAGMA temp_store = MEMORY;
            PRAGMA mmap_size = 30000000000;
            PRAGMA page_size = 4096;
        ")?;

        Ok(Self { conn })
    }
}

4.2 Parameterized Queries (CRITICAL)

// CORRECT: Parameterized query
pub fn get_user_by_id(&self, user_id: i64) -> Result<Option<User>> {
    let mut stmt = self.conn.prepare(
        "SELECT id, name, email FROM users WHERE id = ?1"
    )?;

    let user = stmt.query_row([user_id], |row| {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    }).optional()?;

    Ok(user)
}

// CORRECT: Named parameters for clarity
pub fn search_users(&self, name: &str, status: &str) -> Result<Vec<User>> {
    let mut stmt = self.conn.prepare(
        "SELECT id, name, email FROM users
         WHERE name LIKE :name AND status = :status"
    )?;

    let users = stmt.query_map(
        &[(":name", &format!("%{}%", name)), (":status", &status)],
        |row| Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    )?.collect::<Result<Vec<_>>>()?;

    Ok(users)
}

// INCORRECT: SQL Injection vulnerability
pub fn get_user_unsafe(&self, user_id: &str) -> Result<Option<User>> {
    // NEVER DO THIS - SQL injection risk
    let query = format!("SELECT * FROM users WHERE id = {}", user_id);
    // ...
}

4.3 Transaction Management

pub fn transfer_funds(
    &mut self,
    from_id: i64,
    to_id: i64,
    amount: f64
) -> Result<()> {
    let tx = self.conn.transaction()?;

    // Debit from source
    tx.execute(
        "UPDATE accounts SET balance = balance - ?1 WHERE id = ?2",
        [amount, from_id as f64],
    )?;

    // Credit to destination
    tx.execute(
        "UPDATE accounts SET balance = balance + ?1 WHERE id = ?2",
        [amount, to_id as f64],
    )?;

    tx.commit()?;
    Ok(())
}

4.4 Full-Text Search (FTS5)

// Create FTS5 virtual table with triggers
pub fn setup_fts(&self) -> Result<()> {
    self.conn.execute_batch("
        CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
            title, content, tags, content=documents, content_rowid=id
        );
        CREATE TRIGGER IF NOT EXISTS docs_ai AFTER INSERT ON documents BEGIN
            INSERT INTO docs_fts(rowid, title, content, tags)
            VALUES (new.id, new.title, new.content, new.tags);
        END;
    ")?;
    Ok(())
}

// Search with highlighting
pub fn search_documents(&self, query: &str) -> Result<Vec<Document>> {
    let mut stmt = self.conn.prepare(
        "SELECT d.*, highlight(docs_fts, 1, '<mark>', '</mark>') as snippet
         FROM documents d JOIN docs_fts ON d.id = docs_fts.rowid
         WHERE docs_fts MATCH ?1 ORDER BY rank"
    )?;
    stmt.query_map([query], |row| Ok(Document { /* ... */ }))?.collect()
}

4.5 libSQL / Turso Implementation (TypeScript)

Connection Setup

Local libSQL (Development)

import { createClient } from '@libsql/client'

// Local file database
const db = createClient({
  url: 'file:local.db',
})

// In-memory for testing
const testDb = createClient({
  url: ':memory:',
})

Turso (Production)

import { createClient } from '@libsql/client'

const db = createClient({
  url: process.env.TURSO_DATABASE_URL!,
  authToken: process.env.TURSO_AUTH_TOKEN!,
})

Turso with Embedded Replica (Low Latency)

import { createClient } from '@libsql/client'

// Syncs from remote, reads from local replica
const db = createClient({
  url: 'file:local-replica.db',
  syncUrl: process.env.TURSO_DATABASE_URL!,
  authToken: process.env.TURSO_AUTH_TOKEN!,
  syncInterval: 60, // Sync every 60 seconds
})

// Manual sync when needed
await db.sync()

Parameterized Queries (CRITICAL)

// CORRECT: Parameterized query - ALWAYS use this pattern
async function getUserById(id: string) {
  const result = await db.execute({
    sql: 'SELECT id, name, email FROM users WHERE id = ?',
    args: [id],
  })
  return result.rows[0]
}

// CORRECT: Named parameters for clarity
async function searchUsers(name: string, status: string) {
  const result = await db.execute({
    sql: 'SELECT * FROM users WHERE name LIKE :name AND status = :status',
    args: { name: `%${name}%`, status },
  })
  return result.rows
}

// INCORRECT: SQL Injection vulnerability - NEVER do this
async function getUserUnsafe(id: string) {
  // DANGER: SQL injection risk!
  const result = await db.execute(`SELECT * FROM users WHERE id = '${id}'`)
  return result.rows[0]
}

Batch Operations

// Batch insert with transaction
async function createUsers(users: Array<{ name: string; email: string }>) {
  const statements = users.map((user) => ({
    sql: 'INSERT INTO users (name, email) VALUES (?, ?)',
    args: [user.name, user.email],
  }))

  // Executes all in single round-trip, atomic transaction
  const results = await db.batch(statements, 'write')
  return results
}

Transactions

async function transferFunds(fromId: string, toId: string, amount: number) {
  const tx = await db.transaction('write')

  try {
    await tx.execute({
      sql: 'UPDATE accounts SET balance = balance - ? WHERE id = ?',
      args: [amount, fromId],
    })

    await tx.execute({
      sql: 'UPDATE accounts SET balance = balance + ? WHERE id = ?',
      args: [amount, toId],
    })

    await tx.commit()
  } catch (error) {
    await tx.rollback()
    throw error
  }
}

Repository Pattern (Recommended)

// repositories/user-repository.ts
import { Client } from '@libsql/client'

export class UserRepository {
  constructor(private db: Client) {}

  async findById(id: string) {
    const result = await this.db.execute({
      sql: 'SELECT * FROM users WHERE id = ?',
      args: [id],
    })
    return result.rows[0] ?? null
  }

  async create(data: { name: string; email: string }) {
    const id = crypto.randomUUID()
    await this.db.execute({
      sql: 'INSERT INTO users (id, name, email, created_at) VALUES (?, ?, ?, ?)',
      args: [id, data.name, data.email, new Date().toISOString()],
    })
    return { id, ...data }
  }

  async update(id: string, data: Partial<{ name: string; email: string }>) {
    const sets: string[] = []
    const args: unknown[] = []

    if (data.name !== undefined) {
      sets.push('name = ?')
      args.push(data.name)
    }
    if (data.email !== undefined) {
      sets.push('email = ?')
      args.push(data.email)
    }

    if (sets.length === 0) return

    args.push(id)
    await this.db.execute({
      sql: `UPDATE users SET ${sets.join(', ')} WHERE id = ?`,
      args,
    })
  }

  async delete(id: string) {
    await this.db.execute({
      sql: 'DELETE FROM users WHERE id = ?',
      args: [id],
    })
  }
}

Migrations for libSQL/Turso

// migrations/001_create_users.ts
export const up = async (db: Client) => {
  await db.execute(`
    CREATE TABLE IF NOT EXISTS users (
      id TEXT PRIMARY KEY,
      name TEXT NOT NULL,
      email TEXT NOT NULL UNIQUE,
      created_at TEXT NOT NULL,
      updated_at TEXT
    )
  `)
  await db.execute(`
    CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)
  `)
}

export const down = async (db: Client) => {
  await db.execute('DROP TABLE IF EXISTS users')
}

// migrate.ts - Simple migration runner
import { createClient } from '@libsql/client'
import * as migration001 from './migrations/001_create_users'

const migrations = [migration001]

async function migrate() {
  const db = createClient({ url: process.env.DATABASE_URL! })

  await db.execute(`
    CREATE TABLE IF NOT EXISTS _migrations (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      name TEXT NOT NULL UNIQUE,
      applied_at TEXT NOT NULL
    )
  `)

  const applied = await db.execute('SELECT name FROM _migrations')
  const appliedNames = new Set(applied.rows.map((r) => r.name))

  for (const [index, migration] of migrations.entries()) {
    const name = `${String(index + 1).padStart(3, '0')}`
    if (!appliedNames.has(name)) {
      console.log(`Applying migration ${name}...`)
      await migration.up(db)
      await db.execute({
        sql: 'INSERT INTO _migrations (name, applied_at) VALUES (?, ?)',
        args: [name, new Date().toISOString()],
      })
    }
  }

  console.log('Migrations complete')
}

Edge Deployment Patterns

Cloudflare Workers

import { createClient } from '@libsql/client/web'

export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const db = createClient({
      url: env.TURSO_DATABASE_URL,
      authToken: env.TURSO_AUTH_TOKEN,
    })

    const users = await db.execute('SELECT * FROM users LIMIT 10')
    return Response.json(users.rows)
  },
}

Vercel Edge Functions

import { createClient } from '@libsql/client/web'

export const config = { runtime: 'edge' }

export default async function handler(request: Request) {
  const db = createClient({
    url: process.env.TURSO_DATABASE_URL!,
    authToken: process.env.TURSO_AUTH_TOKEN!,
  })

  const result = await db.execute('SELECT COUNT(*) as count FROM users')
  return Response.json({ count: result.rows[0].count })
}

Bun with libSQL

import { createClient } from '@libsql/client'

const db = createClient({
  url: process.env.TURSO_DATABASE_URL!,
  authToken: process.env.TURSO_AUTH_TOKEN!,
})

Bun.serve({
  port: 3000,
  async fetch(req) {
    const url = new URL(req.url)

    if (url.pathname === '/users') {
      const result = await db.execute('SELECT * FROM users')
      return Response.json(result.rows)
    }

    return new Response('Not found', { status: 404 })
  },
})

5. Security Standards

5.1 Key Vulnerabilities

Mitigation: Update to SQLite 3.44.0+ and always use parameterized queries.

5.2 OWASP Mapping

OWASP Category	Risk	Key Controls
A03 - Injection	Critical	Parameterized queries, input validation
A04 - Insecure Design	Medium	Schema constraints, foreign keys
A05 - Misconfiguration	Medium	Secure PRAGMAs, file permissions (600)

5.3 SQL Injection Prevention

Critical Rules (see references/security-examples.md):

NEVER use string formatting for SQL queries
ALWAYS use ? positional or :name named parameters
Whitelist column/table names for dynamic queries

// Dynamic column selection - SAFE approach
pub fn get_user_fields(&self, user_id: i64, fields: &[&str]) -> Result<HashMap<String, String>> {
    const ALLOWED: &[&str] = &["id", "name", "email", "created_at"];
    let safe_fields: Vec<&str> = fields.iter()
        .filter(|f| ALLOWED.contains(f)).copied().collect();
    if safe_fields.is_empty() { return Err(rusqlite::Error::InvalidQuery); }
    let query = format!("SELECT {} FROM users WHERE id = ?1", safe_fields.join(", "));
    let mut stmt = self.conn.prepare(&query)?;
    // ...
}

6. Testing Standards

6.1 Rust Testing Pattern

#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::Connection;

    fn setup_test_db() -> Database {
        let conn = Connection::open_in_memory().unwrap();
        let db = Database { conn };
        db.run_migrations().unwrap();
        db
    }

    #[test]
    fn test_sql_injection_prevented() {
        let db = setup_test_db();
        let result = db.search_users("'; DROP TABLE users; --", "active");
        assert!(result.is_ok());
        assert!(db.get_user_by_id(1).is_ok()); // Table still exists
    }
}

6.2 TypeScript/libSQL Testing Pattern

import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import { createClient, Client } from '@libsql/client'
import { UserRepository } from './user-repository'

describe('UserRepository', () => {
  let db: Client
  let repo: UserRepository

  beforeEach(async () => {
    // In-memory database for fast, isolated tests
    db = createClient({ url: ':memory:' })

    // Run migrations
    await db.execute(`
      CREATE TABLE users (
        id TEXT PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT NOT NULL UNIQUE,
        created_at TEXT NOT NULL
      )
    `)

    repo = new UserRepository(db)
  })

  afterEach(() => {
    db.close()
  })

  it('creates user with valid data', async () => {
    const user = await repo.create({ name: 'Test', email: 'test@example.com' })

    expect(user.id).toBeDefined()
    expect(user.name).toBe('Test')
  })

  it('finds user by id', async () => {
    const created = await repo.create({ name: 'Test', email: 'test@example.com' })
    const found = await repo.findById(created.id)

    expect(found?.name).toBe('Test')
  })

  it('prevents SQL injection in search', async () => {
    await repo.create({ name: 'Test', email: 'test@example.com' })

    // Malicious input should be safely escaped
    const result = await repo.search("'; DROP TABLE users; --")

    // Table should still exist
    const count = await db.execute('SELECT COUNT(*) as count FROM users')
    expect(count.rows[0].count).toBe(1)
  })

  it('handles unique constraint violation', async () => {
    await repo.create({ name: 'Test', email: 'test@example.com' })

    await expect(repo.create({ name: 'Test 2', email: 'test@example.com' })).rejects.toThrow(
      /UNIQUE constraint failed/
    )
  })
})

// Integration test with Turso (for CI/staging)
describe('UserRepository (Turso integration)', () => {
  let db: Client
  let repo: UserRepository

  beforeEach(async () => {
    // Skip if no Turso credentials
    if (!process.env.TURSO_TEST_URL) {
      return
    }

    db = createClient({
      url: process.env.TURSO_TEST_URL!,
      authToken: process.env.TURSO_TEST_TOKEN!,
    })

    // Clean test data
    await db.execute("DELETE FROM users WHERE email LIKE '%@test.example.com'")
    repo = new UserRepository(db)
  })

  afterEach(async () => {
    if (db) {
      await db.execute("DELETE FROM users WHERE email LIKE '%@test.example.com'")
      db.close()
    }
  })

  it('creates user in Turso', async () => {
    if (!process.env.TURSO_TEST_URL) {
      return // Skip
    }

    const user = await repo.create({
      name: 'Integration Test',
      email: `${Date.now()}@test.example.com`,
    })

    expect(user.id).toBeDefined()
  })
})

7. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# tests/test_user_repository.py
import pytest
import sqlite3

@pytest.fixture
def db():
    """In-memory SQLite for fast testing."""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    yield conn
    conn.close()

class TestUserRepository:
    def test_create_user_returns_id(self, db):
        repo = UserRepository(db)
        repo.initialize_schema()
        user_id = repo.create_user("test@example.com", "Test User")
        assert user_id > 0

    def test_sql_injection_prevented(self, db):
        repo = UserRepository(db)
        repo.initialize_schema()
        malicious = "'; DROP TABLE users; --"
        user_id = repo.create_user(malicious, "Hacker")
        assert repo.get_by_id(user_id)["email"] == malicious

Step 2: Implement Minimum Code to Pass

# app/repositories/user.py
class UserRepository:
    def __init__(self, conn):
        self.conn = conn

    def initialize_schema(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                email TEXT NOT NULL UNIQUE,
                name TEXT NOT NULL
            )""")
        self.conn.commit()

    def create_user(self, email: str, name: str) -> int:
        cursor = self.conn.execute(
            "INSERT INTO users (email, name) VALUES (?, ?)", (email, name))
        self.conn.commit()
        return cursor.lastrowid

    def get_by_id(self, user_id: int):
        return self.conn.execute(
            "SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()

Step 3: Run Verification

pytest tests/test_*_repository.py -v --cov=app/repositories

7.1 Performance Patterns

Pattern 1: WAL Mode

# Good: Enable WAL for concurrent read/write
conn.execute("PRAGMA journal_mode = WAL")
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -64000")  # 64MB

# Bad: Default DELETE mode blocks reads during writes

Pattern 2: Batch Inserts

# Good: Single transaction for batch
conn.executemany("INSERT INTO items (name) VALUES (?)", records)
conn.commit()

# Bad: Commit per row (100x slower)
for r in records:
    conn.execute("INSERT INTO items (name) VALUES (?)", (r,))
    conn.commit()

Pattern 3: Connection Pooling

# Good: Reuse connections
from queue import Queue
class ConnectionPool:
    def __init__(self, db_path, size=5):
        self.pool = Queue(size)
        for _ in range(size):
            conn = sqlite3.connect(db_path, check_same_thread=False)
            conn.execute("PRAGMA journal_mode = WAL")
            self.pool.put(conn)

# Bad: New connection per query
conn = sqlite3.connect(db_path)  # Expensive!

Pattern 4: Index Optimization

# Good: Covering and partial indexes
conn.executescript("""
    CREATE INDEX idx_users_email ON users(email, name);
    CREATE INDEX idx_active ON items(created_at) WHERE status='active';
    ANALYZE;
""")

# Bad: Full table scan on unindexed columns

Pattern 5: VACUUM Scheduling

# Good: Maintenance during idle time
def nightly_maintenance(conn):
    conn.execute("PRAGMA optimize")
    freelist = conn.execute("PRAGMA freelist_count").fetchone()[0]
    if freelist > 1000:
        conn.execute("VACUUM")

# Bad: VACUUM during peak usage or never

8. Common Mistakes

Mistake	Wrong	Correct
SQL Injection	`format!("...WHERE name = '{}'", input)`	`"...WHERE name = ?1"` with params
No Transaction	Separate execute calls	Wrap in `transaction()` + `commit()`
No Foreign Keys	Default connection	`PRAGMA foreign_keys = ON`
LIKE for Search	`LIKE '%term%'`	FTS5 `MATCH 'term'`

13. Pre-Implementation Checklist

Phase 1: Before Writing Code

Tests written first - Create failing tests for new database operations
Schema designed - Document table structure, constraints, indexes
Security reviewed - Identify all user inputs that reach database
Performance targets set - Define query time limits and batch sizes
Reference files read - Load references/security-examples.md if handling user input

Phase 2: During Implementation

Parameterized queries only - Never concatenate user input into SQL
Dynamic names whitelisted - Column/table names from approved list only
Transactions for related ops - Wrap multi-step operations in transactions
Foreign keys enabled - PRAGMA foreign_keys = ON at connection
WAL mode configured - For concurrent read/write access
Indexes created - On columns used in WHERE, JOIN, ORDER BY
Batch operations used - executemany() for multiple inserts
Error handling secure - No SQL details in user-facing errors

Phase 3: Before Committing

All tests pass - Run pytest tests/test_*_repository.py -v
SQL injection test exists - Verify malicious input is safely handled
Performance verified - EXPLAIN QUERY PLAN shows index usage
Migrations tested - Rollback works correctly
Schema version updated - Migration tracking in place
Database permissions set - File mode 600 for production
Backup strategy documented - Recovery procedure verified
VACUUM scheduled - Maintenance plan for database growth

14. Summary

Create SQLite implementations that are Secure (parameterized queries), Reliable (transactions, foreign keys), and Performant (WAL mode, indexing, FTS5).

Security Reminder: NEVER concatenate user input into SQL. ALWAYS use parameterized queries.

SQLite Database Expert