Sqlite Database Expert

SQLite 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 databases in desktop applications handle user data locally, 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 embedded database development, specializing in:

• 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
• Rust/Tauri integration using rusqlite and sea-query

Core Principles

1. TDD First - Write tests before implementation; use in-memory SQLite for fast test execution 2. Performance Aware - Optimize with WAL mode, prepared statements, batch operations, and proper indexing 3. Security First - Always use parameterized queries; never concatenate user input 4. Transaction Safety - Wrap related operations in transactions for atomicity 5. Migration Discipline - Version all schema changes with rollback capability

Primary Use Cases

• Local data persistence for desktop applications
• Offline-first application data storage
• Full-text search implementation
• Configuration and settings storage
• Cache and temporary data management

---

2. Core Responsibilities

2.1 Security-First Database Operations

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

2.2 Data Integrity Principles

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

---

3. Technical Foundation

3.1 Version Recommendations

3.2 Required Dependencies (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"

---

4. Implementation Patterns

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()
}

---

5. Security Standards

5.1 Key Vulnerabilities

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

5.2 OWASP Mapping

5.3 SQL Injection Prevention

Critical Rules (see references/security-examples.md): 1. NEVER use string formatting for SQL queries 2. ALWAYS use ? positional or :name named parameters 3. 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
    }
}

---

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

---

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 Advanced Patterns

Database Migrations

Migration System Implementation

use rusqlite::{Connection, Result, Transaction};

pub struct MigrationManager {
    migrations: Vec<Migration>,
}

struct Migration {
    version: i32,
    name: &'static str,
    up: &'static str,
    down: &'static str,
}

impl MigrationManager {
    pub fn new() -> Self {
        Self {
            migrations: vec![
                Migration {
                    version: 1,
                    name: "create_users_table",
                    up: "CREATE TABLE users (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        name TEXT NOT NULL,
                        email TEXT UNIQUE NOT NULL,
                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
                    )",
                    down: "DROP TABLE users",
                },
                Migration {
                    version: 2,
                    name: "create_documents_table",
                    up: "CREATE TABLE documents (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        user_id INTEGER NOT NULL,
                        title TEXT NOT NULL,
                        content TEXT,
                        tags TEXT,
                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
                        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                    )",
                    down: "DROP TABLE documents",
                },
                Migration {
                    version: 3,
                    name: "add_user_status",
                    up: "ALTER TABLE users ADD COLUMN status TEXT DEFAULT 'active'",
                    down: "ALTER TABLE users DROP COLUMN status",
                },
            ],
        }
    }

    pub fn run(&self, conn: &Connection) -> Result<()> {
        // Create migrations tracking table
        conn.execute(
            "CREATE TABLE IF NOT EXISTS schema_migrations (
                version INTEGER PRIMARY KEY,
                name TEXT NOT NULL,
                applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
            )",
            [],
        )?;

        let current_version: i32 = conn
            .query_row(
                "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
                [],
                |row| row.get(0),
            )
            .unwrap_or(0);

        for migration in &self.migrations {
            if migration.version > current_version {
                println!("Running migration {}: {}", migration.version, migration.name);

                let tx = conn.unchecked_transaction()?;
                tx.execute_batch(migration.up)?;
                tx.execute(
                    "INSERT INTO schema_migrations (version, name) VALUES (?1, ?2)",
                    [&migration.version.to_string(), migration.name],
                )?;
                tx.commit()?;
            }
        }

        Ok(())
    }

    pub fn rollback(&self, conn: &Connection, target_version: i32) -> Result<()> {
        let current_version: i32 = conn
            .query_row(
                "SELECT COALESCE(MAX(version), 0) FROM schema_migrations",
                [],
                |row| row.get(0),
            )?;

        for migration in self.migrations.iter().rev() {
            if migration.version > target_version && migration.version <= current_version {
                println!("Rolling back migration {}: {}", migration.version, migration.name);

                let tx = conn.unchecked_transaction()?;
                tx.execute_batch(migration.down)?;
                tx.execute(
                    "DELETE FROM schema_migrations WHERE version = ?1",
                    [migration.version],
                )?;
                tx.commit()?;
            }
        }

        Ok(())
    }
}

Safe Migration Patterns

// Adding a column with default (safe)
"ALTER TABLE users ADD COLUMN verified BOOLEAN DEFAULT 0"

// Creating an index concurrently is not supported in SQLite
// But you can create indexes normally - SQLite handles this well
"CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)"

// Renaming a table
"ALTER TABLE old_name RENAME TO new_name"

// Copying data to new schema (for complex changes)
pub fn migrate_complex_schema_change(conn: &Connection) -> Result<()> {
    let tx = conn.transaction()?;

    // Create new table with desired schema
    tx.execute_batch("
        CREATE TABLE users_new (
            id INTEGER PRIMARY KEY,
            full_name TEXT NOT NULL,  -- renamed from 'name'
            email TEXT UNIQUE NOT NULL,
            status TEXT DEFAULT 'active',
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        );

        -- Copy data with transformation
        INSERT INTO users_new (id, full_name, email, status, created_at)
        SELECT id, name, email, COALESCE(status, 'active'), created_at
        FROM users;

        -- Swap tables
        DROP TABLE users;
        ALTER TABLE users_new RENAME TO users;
    ")?;

    tx.commit()
}

---

Full-Text Search (FTS5) Advanced

Complex FTS5 Setup

pub fn setup_advanced_fts(conn: &Connection) -> Result<()> {
    conn.execute_batch("
        -- Create FTS5 table with advanced options
        CREATE VIRTUAL TABLE IF NOT EXISTS docs_fts USING fts5(
            title,
            content,
            tags,
            content=documents,
            content_rowid=id,
            tokenize='porter unicode61 remove_diacritics 2',
            prefix='2,3'
        );

        -- Create triggers for sync
        CREATE TRIGGER IF NOT EXISTS docs_fts_insert AFTER INSERT ON documents BEGIN
            INSERT INTO docs_fts(rowid, title, content, tags)
            VALUES (new.id, new.title, new.content, new.tags);
        END;

        CREATE TRIGGER IF NOT EXISTS docs_fts_delete AFTER DELETE ON documents BEGIN
            INSERT INTO docs_fts(docs_fts, rowid, title, content, tags)
            VALUES ('delete', old.id, old.title, old.content, old.tags);
        END;

        CREATE TRIGGER IF NOT EXISTS docs_fts_update AFTER UPDATE ON documents BEGIN
            INSERT INTO docs_fts(docs_fts, rowid, title, content, tags)
            VALUES ('delete', old.id, old.title, old.content, old.tags);
            INSERT INTO docs_fts(rowid, title, content, tags)
            VALUES (new.id, new.title, new.content, new.tags);
        END;
    ")?;
    Ok(())
}

Advanced FTS5 Queries

// Boolean search with operators
pub fn search_boolean(&self, query: &str) -> Result<Vec<Document>> {
    // FTS5 supports: AND, OR, NOT, NEAR, phrase queries
    let mut stmt = self.conn.prepare(
        "SELECT d.*, bm25(docs_fts) as relevance
         FROM documents d
         JOIN docs_fts ON d.id = docs_fts.rowid
         WHERE docs_fts MATCH ?1
         ORDER BY relevance"
    )?;

    // Example queries:
    // "rust AND async" - both terms
    // "rust OR go" - either term
    // "rust NOT unsafe" - exclude term
    // "NEAR(rust async)" - terms near each other
    // '"exact phrase"' - exact phrase match
    // "rust*" - prefix match

    stmt.query_map([query], |row| {
        Ok(Document {
            id: row.get(0)?,
            title: row.get(1)?,
            content: row.get(2)?,
            tags: row.get(3)?,
            relevance: row.get("relevance")?,
        })
    })?.collect()
}

// Column-specific search
pub fn search_by_title(&self, query: &str) -> Result<Vec<Document>> {
    let mut stmt = self.conn.prepare(
        "SELECT d.*
         FROM documents d
         JOIN docs_fts ON d.id = docs_fts.rowid
         WHERE docs_fts MATCH 'title:' || ?1
         ORDER BY rank"
    )?;

    stmt.query_map([query], |row| {
        Ok(Document { ... })
    })?.collect()
}

// Highlight and snippet
pub fn search_with_snippets(&self, query: &str) -> Result<Vec<SearchResult>> {
    let mut stmt = self.conn.prepare(
        "SELECT d.id, d.title,
                snippet(docs_fts, 1, '<b>', '</b>', '...', 32) as content_snippet,
                highlight(docs_fts, 0, '<mark>', '</mark>') as title_highlighted
         FROM documents d
         JOIN docs_fts ON d.id = docs_fts.rowid
         WHERE docs_fts MATCH ?1
         ORDER BY bm25(docs_fts)
         LIMIT 20"
    )?;

    stmt.query_map([query], |row| {
        Ok(SearchResult {
            id: row.get(0)?,
            title: row.get(1)?,
            snippet: row.get(2)?,
            highlighted_title: row.get(3)?,
        })
    })?.collect()
}

FTS5 Maintenance

// Rebuild FTS index
pub fn rebuild_fts_index(conn: &Connection) -> Result<()> {
    conn.execute("INSERT INTO docs_fts(docs_fts) VALUES('rebuild')", [])?;
    Ok(())
}

// Optimize FTS index
pub fn optimize_fts_index(conn: &Connection) -> Result<()> {
    conn.execute("INSERT INTO docs_fts(docs_fts) VALUES('optimize')", [])?;
    Ok(())
}

// Integrity check
pub fn check_fts_integrity(conn: &Connection) -> Result<bool> {
    let result: String = conn.query_row(
        "INSERT INTO docs_fts(docs_fts) VALUES('integrity-check')",
        [],
        |row| row.get(0),
    ).unwrap_or_else(|_| "ok".to_string());

    Ok(result == "ok")
}

---

Connection Pooling

R2D2 Connection Pool Setup

use r2d2::{Pool, PooledConnection};
use r2d2_sqlite::SqliteConnectionManager;
use rusqlite::OpenFlags;
use std::path::Path;
use std::sync::Arc;

pub struct DatabasePool {
    pool: Arc<Pool<SqliteConnectionManager>>,
}

impl DatabasePool {
    pub fn new(path: &Path, pool_size: u32) -> Result<Self, r2d2::Error> {
        let manager = SqliteConnectionManager::file(path)
            .with_flags(
                OpenFlags::SQLITE_OPEN_READ_WRITE
                | OpenFlags::SQLITE_OPEN_CREATE
                | OpenFlags::SQLITE_OPEN_NO_MUTEX
            )
            .with_init(|conn| {
                conn.execute_batch("
                    PRAGMA foreign_keys = ON;
                    PRAGMA journal_mode = WAL;
                    PRAGMA synchronous = NORMAL;
                    PRAGMA busy_timeout = 5000;
                    PRAGMA temp_store = MEMORY;
                    PRAGMA cache_size = -64000;
                ")
            });

        let pool = Pool::builder()
            .max_size(pool_size)
            .min_idle(Some(2))
            .build(manager)?;

        Ok(Self {
            pool: Arc::new(pool),
        })
    }

    pub fn get(&self) -> Result<PooledConnection<SqliteConnectionManager>, r2d2::Error> {
        self.pool.get()
    }

    pub fn state(&self) -> r2d2::State {
        self.pool.state()
    }
}

// Usage in Tauri commands
#[tauri::command]
pub async fn get_users(
    pool: tauri::State<'_, DatabasePool>
) -> Result<Vec<User>, String> {
    let conn = pool.get().map_err(|e| e.to_string())?;

    let mut stmt = conn.prepare(
        "SELECT id, name, email FROM users"
    ).map_err(|e| e.to_string())?;

    let users = stmt.query_map([], |row| {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    })
    .map_err(|e| e.to_string())?
    .collect::<Result<Vec<_>, _>>()
    .map_err(|e| e.to_string())?;

    Ok(users)
}

---

Advanced Query Patterns

Common Table Expressions (CTEs)

// Recursive CTE for hierarchical data
pub fn get_category_tree(&self, root_id: i64) -> Result<Vec<Category>> {
    let mut stmt = self.conn.prepare(
        "WITH RECURSIVE category_tree AS (
            -- Base case: root category
            SELECT id, name, parent_id, 0 as depth
            FROM categories
            WHERE id = ?1

            UNION ALL

            -- Recursive case: children
            SELECT c.id, c.name, c.parent_id, ct.depth + 1
            FROM categories c
            JOIN category_tree ct ON c.parent_id = ct.id
        )
        SELECT * FROM category_tree ORDER BY depth, name"
    )?;

    stmt.query_map([root_id], |row| {
        Ok(Category {
            id: row.get(0)?,
            name: row.get(1)?,
            parent_id: row.get(2)?,
            depth: row.get(3)?,
        })
    })?.collect()
}

Window Functions

// Ranking and analytics
pub fn get_user_rankings(&self) -> Result<Vec<UserRanking>> {
    let mut stmt = self.conn.prepare(
        "SELECT
            id,
            name,
            score,
            RANK() OVER (ORDER BY score DESC) as rank,
            DENSE_RANK() OVER (ORDER BY score DESC) as dense_rank,
            ROW_NUMBER() OVER (ORDER BY score DESC) as row_num,
            score - LAG(score) OVER (ORDER BY score DESC) as score_diff,
            SUM(score) OVER (ORDER BY score DESC ROWS UNBOUNDED PRECEDING) as cumulative_score
         FROM users
         ORDER BY rank"
    )?;

    stmt.query_map([], |row| {
        Ok(UserRanking {
            id: row.get(0)?,
            name: row.get(1)?,
            score: row.get(2)?,
            rank: row.get(3)?,
            dense_rank: row.get(4)?,
            row_num: row.get(5)?,
            score_diff: row.get(6)?,
            cumulative_score: row.get(7)?,
        })
    })?.collect()
}

JSON Functions

// Working with JSON data
pub fn query_json_field(&self, field_path: &str, value: &str) -> Result<Vec<Record>> {
    let mut stmt = self.conn.prepare(
        "SELECT id, data
         FROM records
         WHERE json_extract(data, ?1) = ?2"
    )?;

    stmt.query_map([field_path, value], |row| {
        Ok(Record {
            id: row.get(0)?,
            data: row.get(1)?,
        })
    })?.collect()
}

// Aggregating JSON arrays
pub fn aggregate_tags(&self) -> Result<Vec<String>> {
    let mut stmt = self.conn.prepare(
        "SELECT DISTINCT json_each.value
         FROM documents, json_each(documents.tags)
         ORDER BY json_each.value"
    )?;

    stmt.query_map([], |row| row.get(0))?.collect()
}

---

Performance Optimization

Index Strategies

-- Single column index
CREATE INDEX idx_users_email ON users(email);

-- Composite index (order matters!)
CREATE INDEX idx_docs_user_created ON documents(user_id, created_at DESC);

-- Partial index (for filtered queries)
CREATE INDEX idx_active_users ON users(email) WHERE status = 'active';

-- Covering index (includes all needed columns)
CREATE INDEX idx_docs_cover ON documents(user_id, created_at, title);

-- Expression index
CREATE INDEX idx_users_lower_email ON users(LOWER(email));

Query Analysis

pub fn analyze_query(conn: &Connection, query: &str) -> Result<String> {
    let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {}", query))?;
    let plan: Vec<String> = stmt.query_map([], |row| {
        Ok(format!("{}: {}", row.get::<_, i32>(0)?, row.get::<_, String>(3)?))
    })?.collect::<Result<Vec<_>, _>>()?;

    Ok(plan.join("\n"))
}

VACUUM and Maintenance

pub fn maintain_database(conn: &Connection) -> Result<()> {
    // Analyze for query planner
    conn.execute("ANALYZE", [])?;

    // Reclaim space (only when needed - this rewrites entire database)
    conn.execute("VACUUM", [])?;

    // Optimize FTS
    conn.execute("INSERT INTO docs_fts(docs_fts) VALUES('optimize')", [])?;

    Ok(())
}

SQLite Security Examples

SQL Injection Prevention

Parameterized Query Patterns

use rusqlite::{Connection, Result, params};

// CORRECT: Positional parameters
pub fn get_user_by_id(conn: &Connection, id: i64) -> Result<Option<User>> {
    conn.query_row(
        "SELECT id, name, email FROM users WHERE id = ?1",
        [id],
        |row| Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    ).optional()
}

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

    stmt.query_map(
        &[(":name", &format!("%{}%", escape_like_pattern(name))),
          (":status", &status.to_string())],
        |row| Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    )?.collect()
}

// CORRECT: Using params! macro for multiple types
pub fn insert_document(
    conn: &Connection,
    user_id: i64,
    title: &str,
    content: &str
) -> Result<i64> {
    conn.execute(
        "INSERT INTO documents (user_id, title, content) VALUES (?1, ?2, ?3)",
        params![user_id, title, content],
    )?;
    Ok(conn.last_insert_rowid())
}

LIKE Pattern Escaping

/// Escape special characters in LIKE patterns to prevent injection
pub fn escape_like_pattern(pattern: &str) -> String {
    pattern
        .replace('\\', "\\\\")
        .replace('%', "\\%")
        .replace('_', "\\_")
}

pub fn search_by_name(conn: &Connection, search: &str) -> Result<Vec<User>> {
    let escaped = escape_like_pattern(search);
    let pattern = format!("%{}%", escaped);

    let mut stmt = conn.prepare(
        "SELECT id, name, email FROM users WHERE name LIKE ?1 ESCAPE '\\'"
    )?;

    stmt.query_map([pattern], |row| {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    })?.collect()
}

Dynamic Query Building with Whitelisting

/// Build dynamic queries safely with whitelisted columns
pub fn get_users_sorted(
    conn: &Connection,
    sort_column: &str,
    sort_order: &str
) -> Result<Vec<User>> {
    // Whitelist allowed columns
    const ALLOWED_COLUMNS: &[&str] = &["id", "name", "email", "created_at"];
    const ALLOWED_ORDERS: &[&str] = &["ASC", "DESC"];

    let column = ALLOWED_COLUMNS
        .iter()
        .find(|&&c| c == sort_column)
        .ok_or(rusqlite::Error::InvalidParameterName("Invalid sort column".into()))?;

    let order = ALLOWED_ORDERS
        .iter()
        .find(|&&o| o == sort_order.to_uppercase())
        .unwrap_or(&"ASC");

    // Safe to interpolate - values are whitelisted
    let query = format!(
        "SELECT id, name, email FROM users ORDER BY {} {}",
        column, order
    );

    let mut stmt = conn.prepare(&query)?;
    stmt.query_map([], |row| {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    })?.collect()
}

/// Dynamic column selection with whitelisting
pub fn get_user_fields(
    conn: &Connection,
    user_id: i64,
    fields: &[String]
) -> Result<serde_json::Value> {
    const ALLOWED: &[&str] = &["id", "name", "email", "status", "created_at"];

    let safe_fields: Vec<&str> = fields
        .iter()
        .filter_map(|f| ALLOWED.iter().find(|&&a| a == f.as_str()).copied())
        .collect();

    if safe_fields.is_empty() {
        return Err(rusqlite::Error::InvalidQuery);
    }

    let columns = safe_fields.join(", ");
    let query = format!("SELECT {} FROM users WHERE id = ?1", columns);

    let mut stmt = conn.prepare(&query)?;
    let result = stmt.query_row([user_id], |row| {
        let mut map = serde_json::Map::new();
        for (i, field) in safe_fields.iter().enumerate() {
            let value: rusqlite::types::Value = row.get(i)?;
            map.insert(
                field.to_string(),
                match value {
                    rusqlite::types::Value::Null => serde_json::Value::Null,
                    rusqlite::types::Value::Integer(i) => serde_json::json!(i),
                    rusqlite::types::Value::Real(f) => serde_json::json!(f),
                    rusqlite::types::Value::Text(s) => serde_json::json!(s),
                    rusqlite::types::Value::Blob(b) => serde_json::json!(base64::encode(b)),
                }
            );
        }
        Ok(serde_json::Value::Object(map))
    })?;

    Ok(result)
}

Using Query Builder (sea-query)

use sea_query::{Expr, Query, SqliteQueryBuilder, Iden};
use sea_query_rusqlite::RusqliteBinder;

#[derive(Iden)]
enum Users {
    Table,
    Id,
    Name,
    Email,
    Status,
}

pub fn build_user_search(
    conn: &Connection,
    name: Option<&str>,
    status: Option<&str>,
    limit: u64
) -> Result<Vec<User>> {
    let mut query = Query::select();
    query
        .columns([Users::Id, Users::Name, Users::Email])
        .from(Users::Table);

    if let Some(name) = name {
        query.and_where(Expr::col(Users::Name).like(format!("%{}%", name)));
    }

    if let Some(status) = status {
        query.and_where(Expr::col(Users::Status).eq(status));
    }

    query.limit(limit);

    let (sql, values) = query.build_rusqlite(SqliteQueryBuilder);

    let mut stmt = conn.prepare(&sql)?;
    let params = values.as_params();

    stmt.query_map(&*params, |row| {
        Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    })?.collect()
}

---

Input Validation

Multi-Layer Validation

use serde::{Deserialize, Serialize};
use validator::{Validate, ValidationError};

#[derive(Debug, Deserialize, Validate)]
pub struct CreateUserRequest {
    #[validate(length(min = 2, max = 50))]
    pub name: String,

    #[validate(email)]
    pub email: String,

    #[validate(length(min = 12), custom = "validate_password_strength")]
    pub password: String,
}

fn validate_password_strength(password: &str) -> Result<(), ValidationError> {
    let has_upper = password.chars().any(|c| c.is_uppercase());
    let has_lower = password.chars().any(|c| c.is_lowercase());
    let has_digit = password.chars().any(|c| c.is_numeric());
    let has_special = password.chars().any(|c| !c.is_alphanumeric());

    if has_upper && has_lower && has_digit && has_special {
        Ok(())
    } else {
        Err(ValidationError::new("weak_password"))
    }
}

pub fn create_user(conn: &Connection, request: CreateUserRequest) -> Result<User> {
    // Layer 1: Schema validation
    request.validate().map_err(|e| {
        rusqlite::Error::InvalidParameterName(format!("Validation failed: {}", e))
    })?;

    // Layer 2: Business validation
    let email_exists: bool = conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM users WHERE email = ?1)",
        [&request.email],
        |row| row.get(0)
    )?;

    if email_exists {
        return Err(rusqlite::Error::InvalidParameterName(
            "Email already exists".into()
        ));
    }

    // Layer 3: Hash password before storage
    let password_hash = hash_password(&request.password)?;

    // Layer 4: Parameterized insert
    conn.execute(
        "INSERT INTO users (name, email, password_hash) VALUES (?1, ?2, ?3)",
        params![request.name, request.email, password_hash],
    )?;

    Ok(User {
        id: conn.last_insert_rowid(),
        name: request.name,
        email: request.email,
    })
}

Integer Parsing (Not String Interpolation)

// WRONG: String interpolation of user input
pub fn get_user_unsafe(conn: &Connection, id_str: &str) -> Result<User> {
    let query = format!("SELECT * FROM users WHERE id = {}", id_str);
    // SQL Injection if id_str = "1; DROP TABLE users; --"
}

// CORRECT: Parse and use as typed parameter
pub fn get_user(conn: &Connection, id_str: &str) -> Result<User> {
    let id: i64 = id_str.parse().map_err(|_| {
        rusqlite::Error::InvalidParameterName("Invalid user ID".into())
    })?;

    conn.query_row(
        "SELECT id, name, email FROM users WHERE id = ?1",
        [id],
        |row| Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    )
}

---

Error Handling

Safe Error Messages

use thiserror::Error;

#[derive(Error, Debug)]
pub enum DatabaseError {
    #[error("User not found")]
    NotFound,

    #[error("Invalid input: {0}")]
    Validation(String),

    #[error("Email already exists")]
    DuplicateEmail,

    #[error("Database operation failed")]
    Internal(#[from] rusqlite::Error),
}

// Convert internal errors to safe user-facing messages
impl From<DatabaseError> for String {
    fn from(err: DatabaseError) -> Self {
        match err {
            DatabaseError::NotFound => "Resource not found".into(),
            DatabaseError::Validation(msg) => msg,
            DatabaseError::DuplicateEmail => "Email already in use".into(),
            // Don't expose internal database errors
            DatabaseError::Internal(e) => {
                // Log the actual error internally
                log::error!("Database error: {:?}", e);
                "An internal error occurred".into()
            }
        }
    }
}

pub fn get_user_safe(conn: &Connection, id: i64) -> Result<User, DatabaseError> {
    conn.query_row(
        "SELECT id, name, email FROM users WHERE id = ?1",
        [id],
        |row| Ok(User {
            id: row.get(0)?,
            name: row.get(1)?,
            email: row.get(2)?,
        })
    ).optional()?
      .ok_or(DatabaseError::NotFound)
}

---

File Permissions & Security

Database File Security

use std::fs;
use std::os::unix::fs::PermissionsExt;

pub fn secure_database_file(path: &std::path::Path) -> std::io::Result<()> {
    // Set restrictive permissions (owner read/write only)
    let permissions = fs::Permissions::from_mode(0o600);
    fs::set_permissions(path, permissions)?;

    // Also secure the WAL and SHM files if they exist
    let wal_path = path.with_extension("db-wal");
    let shm_path = path.with_extension("db-shm");

    if wal_path.exists() {
        fs::set_permissions(&wal_path, fs::Permissions::from_mode(0o600))?;
    }
    if shm_path.exists() {
        fs::set_permissions(&shm_path, fs::Permissions::from_mode(0o600))?;
    }

    Ok(())
}

// Platform-independent path security
#[cfg(target_os = "windows")]
pub fn secure_database_file(path: &std::path::Path) -> std::io::Result<()> {
    // On Windows, use ACLs via windows-acl crate or similar
    // This is a simplified example
    Ok(())
}

Secure Database Initialization

use std::path::PathBuf;
use directories::ProjectDirs;

pub fn get_secure_db_path() -> Result<PathBuf, Box<dyn std::error::Error>> {
    let proj_dirs = ProjectDirs::from("com", "yourcompany", "yourapp")
        .ok_or("Failed to get project directories")?;

    let data_dir = proj_dirs.data_local_dir();

    // Create directory with proper permissions
    std::fs::create_dir_all(data_dir)?;

    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(data_dir, std::fs::Permissions::from_mode(0o700))?;
    }

    Ok(data_dir.join("app.db"))
}

---

Backup Security

Encrypted Backups

use rusqlite::backup::{Backup, Progress};
use aes_gcm::{Aes256Gcm, Key, Nonce};
use aes_gcm::aead::{Aead, NewAead};
use rand::Rng;

pub fn create_encrypted_backup(
    conn: &Connection,
    backup_path: &std::path::Path,
    encryption_key: &[u8; 32]
) -> Result<()> {
    // First create unencrypted backup in memory
    let backup_conn = Connection::open_in_memory()?;

    {
        let backup = Backup::new(conn, &backup_conn)?;
        backup.run_to_completion(5, std::time::Duration::from_millis(250), None)?;
    }

    // Export to bytes
    let mut backup_data = Vec::new();
    backup_conn.execute("VACUUM INTO ?1", [":memory:"])?;

    // Serialize the backup
    let mut stmt = backup_conn.prepare("SELECT * FROM sqlite_master")?;
    // ... serialize database

    // Encrypt the backup
    let cipher = Aes256Gcm::new(Key::from_slice(encryption_key));
    let nonce_bytes: [u8; 12] = rand::thread_rng().gen();
    let nonce = Nonce::from_slice(&nonce_bytes);

    let encrypted = cipher.encrypt(nonce, backup_data.as_ref())
        .map_err(|_| rusqlite::Error::InvalidQuery)?;

    // Write nonce + encrypted data
    let mut output = Vec::with_capacity(12 + encrypted.len());
    output.extend_from_slice(&nonce_bytes);
    output.extend(encrypted);

    std::fs::write(backup_path, output)?;

    // Set restrictive permissions
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(backup_path, std::fs::Permissions::from_mode(0o600))?;
    }

    Ok(())
}

---

Audit Logging

Database Operation Logging

use chrono::Utc;
use log::info;

pub struct AuditedConnection {
    conn: Connection,
    user_id: Option<i64>,
}

impl AuditedConnection {
    pub fn execute_audited(
        &self,
        query: &str,
        params: &[&dyn rusqlite::ToSql],
        operation: &str,
        resource: &str
    ) -> Result<usize> {
        let result = self.conn.execute(query, params)?;

        // Log the operation (without sensitive data)
        info!(
            target: "audit",
            "user_id={:?} operation={} resource={} affected_rows={}",
            self.user_id, operation, resource, result
        );

        // Optionally store in audit table
        self.conn.execute(
            "INSERT INTO audit_log (user_id, operation, resource, affected_rows, timestamp)
             VALUES (?1, ?2, ?3, ?4, ?5)",
            params![
                self.user_id,
                operation,
                resource,
                result as i64,
                Utc::now().to_rfc3339()
            ],
        )?;

        Ok(result)
    }
}

---

Security Testing

SQL Injection Test Cases

#[cfg(test)]
mod security_tests {
    use super::*;

    #[test]
    fn test_sql_injection_in_name_search() {
        let db = setup_test_db();

        // Test various SQL injection payloads
        let payloads = vec![
            "'; DROP TABLE users; --",
            "admin' OR '1'='1",
            "' UNION SELECT * FROM users WHERE '1'='1",
            "1; DELETE FROM users WHERE 1=1; --",
            "' OR 1=1--",
            "admin'--",
            "'; INSERT INTO users (name) VALUES ('hacked'); --",
        ];

        for payload in payloads {
            let result = db.search_users(payload, "active");
            assert!(result.is_ok(), "Query should not fail for payload: {}", payload);

            // Verify data integrity
            let count: i64 = db.conn.query_row(
                "SELECT COUNT(*) FROM users",
                [],
                |row| row.get(0)
            ).unwrap();
            assert!(count >= 0, "Table should still exist after payload: {}", payload);
        }
    }

    #[test]
    fn test_integer_id_validation() {
        let db = setup_test_db();

        // These should fail gracefully
        let invalid_ids = vec![
            "abc",
            "1.5",
            "1; DROP TABLE users",
            "-0",
            "9999999999999999999999999",
        ];

        for id in invalid_ids {
            let result = db.get_user_by_string_id(id);
            // Should either return error or None, not panic
            assert!(result.is_err() || result.unwrap().is_none());
        }
    }

    #[test]
    fn test_like_pattern_escaping() {
        let db = setup_test_db();

        // Create user with special characters in name
        db.create_user("test%user", "test@example.com").unwrap();

        // Search should match exactly, not all users
        let results = db.search_users("%", "active").unwrap();
        // Should find only users with literal % in name
        assert!(results.iter().all(|u| u.name.contains('%')));
    }
}