Struct TxStore 

pub struct TxStore<'a> { /* private fields */ }

A DataStore that executes against a pre-held SQLite connection for the duration of a single mutation handler.

Safety contract

rusqlite::Connection is Send but not Sync (it uses RefCells internally for statement caching). The DataStore trait requires Send + Sync, but &'a Connection is neither.

We hand-implement both via unsafe impl because:

1. Construction. TxStore::new is only ever called by FnOpsImpl::call for mutations, after acquiring the runtime’s write lock. The &Connection originates from a MutexGuard that the constructing thread holds.

2. Lifetime. The 'a lifetime ties the TxStore to that guard. The compiler enforces that the TxStore cannot outlive the held lock; it must be dropped before the guard is.

3. Single-threaded use. FnRunner::call() runs the handler synchronously on the calling thread and never spawns threads holding a reference to the TxStore. The Send + Sync bounds on the DataStore trait are satisfied vacuously — no thread other than the caller ever sees this TxStore.

4. No interior aliasing. All &Connection calls go through Runtime::*_with_conn methods which take &Connection, never keeping the reference alive across an await point (this is sync code, no awaits).

Future work: refactor Runtime’s write_conn to be Arc<Mutex<Connection>> so TxStore can hold an Arc<Mutex<...>>, eliminating the unsafe impl entirely.

Implementations

impl<'a> TxStore<'a>

pub fn new(runtime: &'a Runtime, conn: &'a Connection) -> Self

pub fn take_pending(&self) -> Vec<ChangeEvent>

Drain the pending-events buffer. Called after COMMIT succeeds; the caller is responsible for appending each event to the ChangeLog and broadcasting via the notifier. On rollback the caller just drops the buffer without calling this.

Trait Implementations

impl<'a> DataStore for TxStore<'a>

fn manifest(&self) -> &AppManifest

fn insert(&self, entity: &str, data: &Value) -> Result<String, DataError>

fn get_by_id(&self, entity: &str, id: &str) -> Result<Option<Value>, DataError>

fn list(&self, entity: &str) -> Result<Vec<Value>, DataError>

fn list_after( &self, entity: &str, after: Option<&str>, limit: usize, ) -> Result<Vec<Value>, DataError>

fn update( &self, entity: &str, id: &str, data: &Value, ) -> Result<bool, DataError>

fn delete(&self, entity: &str, id: &str) -> Result<bool, DataError>

fn lookup( &self, entity: &str, field: &str, value: &str, ) -> Result<Option<Value>, DataError>

fn link( &self, entity: &str, id: &str, relation: &str, target_id: &str, ) -> Result<bool, DataError>

fn unlink( &self, entity: &str, id: &str, relation: &str, ) -> Result<bool, DataError>

fn query_filtered( &self, entity: &str, filter: &Value, ) -> Result<Vec<Value>, DataError>

fn query_graph(&self, query: &Value) -> Result<Value, DataError>

fn aggregate(&self, entity: &str, spec: &Value) -> Result<Value, DataError>

Run an aggregation query.

fn transact(&self, _ops: &[Value]) -> Result<(bool, Vec<Value>), DataError>

Execute transactional operations. Each element is a JSON object with op (“insert”/“update”/“delete”), entity, and optionally id/data.

fn search(&self, entity: &str, query: &Value) -> Result<Value, DataError>

Run a faceted full-text search against a searchable entity. query is a JSON object with the keys defined by SearchQuery in pylon_storage::search; returns a JSON object shaped like SearchResult ({ hits, facetCounts, total, tookMs }).

fn crdt_snapshot( &self, _entity: &str, _row_id: &str, ) -> Result<Option<Vec<u8>>, DataError>

Return the binary CRDT snapshot for a row, used by the router to ship a binary update over WebSocket after every successful write.

fn crdt_apply_update( &self, _entity: &str, _row_id: &str, _update: &[u8], ) -> Result<Vec<u8>, DataError>

Apply a binary CRDT update from a client to the row’s LoroDoc, project the new state into the SQLite materialized view, and return the post-merge snapshot bytes (so the caller can broadcast them to OTHER subscribed clients).

impl<'a> Send for TxStore<'a>

impl<'a> Sync for TxStore<'a>