Struct SqliteProjectionStore 

pub struct SqliteProjectionStore { /* private fields */ }

SQLite-backed projection store

Uses separate connections for reads and writes. SQLite WAL mode allows readers and writers to operate concurrently at the database level. While each connection still needs mutex protection (rusqlite::Connection is not Sync), using separate connections means reads don’t block writes and vice versa.

Optionally supports a read connection pool for higher concurrency.

Implementations

impl SqliteProjectionStore

pub fn conn(&self) -> &Arc<Mutex<Connection>>

Get the underlying write connection (for migrations and custom queries)

Returns an Arc to the Mutex-protected SQLite connection. Users should lock the mutex to access the connection.

pub async fn query_async<F, R>(&self, f: F) -> Result<R>

where F: FnOnce(&Connection) -> Result<R> + Send + 'static, R: Send + 'static,

Execute a read-only SQL query asynchronously

This method runs the query on a separate thread to avoid blocking, making it safe to call from async contexts. Uses a read-only connection that doesn’t block writes.

Example

let balance: i64 = store.query_async(|conn| {
    conn.query_row("SELECT balance FROM accounts WHERE id = ?1", [account_id], |row| row.get(0))
}).await?;

pub fn query<F, R>(&self, f: F) -> Result<R>

where F: FnOnce(&Connection) -> Result<R>,

Execute a read-only SQL query synchronously

This is a convenience method for non-async contexts. For async contexts, prefer query_async. Uses a read-only connection that doesn’t block writes.

Example

let balance: i64 = store.query(|conn| {
    conn.query_row("SELECT balance FROM accounts WHERE id = ?1", [account_id], |row| row.get(0))
})?;

pub async fn execute_async<F>(&self, f: F) -> Result<()>

where F: FnOnce(&Connection) -> Result<()> + Send + 'static,

Execute arbitrary SQL statements (DDL/DML) asynchronously

Useful for creating tables, indexes, or performing bulk updates. Uses the write connection.

Example

store.execute_async(|conn| {
    conn.execute("CREATE TABLE IF NOT EXISTS balances (id INTEGER PRIMARY KEY, amount INTEGER)", [])?;
    conn.execute("CREATE INDEX IF NOT EXISTS idx_amount ON balances(amount)", [])?;
    Ok(())
}).await?;

pub fn execute<F>(&self, f: F) -> Result<()>

where F: FnOnce(&Connection) -> Result<()>,

Execute arbitrary SQL statements (DDL/DML) synchronously

Uses the write connection.

Example

store.execute(|conn| {
    conn.execute("CREATE TABLE IF NOT EXISTS balances (id INTEGER PRIMARY KEY, amount INTEGER)", [])?;
    Ok(())
})?;

pub fn transaction<F>(&self, f: F) -> Result<()>

where F: FnOnce(&Transaction<'_>) -> Result<()>,

Execute a transaction with multiple SQL statements

The closure receives a transaction object and can execute multiple statements atomically. If the closure returns an error, the transaction is rolled back. Uses the write connection.

Example

store.transaction(|tx| {
    tx.execute("INSERT INTO accounts (id, balance) VALUES (?1, ?2)", params![1, 100])?;
    tx.execute("INSERT INTO accounts (id, balance) VALUES (?1, ?2)", params![2, 200])?;
    Ok(())
})?;

pub async fn transaction_async<F>(&self, f: F) -> Result<()>

where F: FnOnce(&Transaction<'_>) -> Result<()> + Send + 'static,

Execute a transaction asynchronously

Uses the write connection.

pub fn read_pool(&self) -> Option<&Arc<SqliteReadPool>>

Get reference to the read pool (if enabled)

Returns None if read pooling was not enabled in config.

pub fn has_read_pool(&self) -> bool

Check if read pooling is enabled

pub fn db_path(&self) -> &Path

Get the database path

Trait Implementations

impl ProjectionStore for SqliteProjectionStore

type Txn<'a> = SimpleProjectionTxn<'a>

fn open(cfg: ProjectionConfig) -> Result<Self>

Open a projection store

fn close(&self) -> Result<()>

Close the store

fn begin_txn(&self) -> Result<Self::Txn<'_>>

Begin a transaction

fn get_cursor(&self) -> Result<EventId>

Get the current cursor (last applied event ID)

fn migrate(&self, target_version: u32) -> Result<()>

Run migrations to target schema version

fn backup_to(&self, path: &Path) -> Result<()>

Create a backup

fn restore_from(path: &Path, cfg: ProjectionConfig) -> Result<Self>

Restore from a backup

fn schema_version(&self) -> Result<u32>

Get the schema version