oxisql_sqlite_compat/connection.rs
1//! [`SqliteConnection`] — Limbo-backed implementation of [`oxisql_core::Connection`].
2//!
3//! # Concurrency model
4//!
5//! `limbo::Connection` is internally `Arc<Mutex<Arc<limbo_core::Connection>>>` with
6//! `unsafe impl Send + Sync`, so it is safe to clone and share across async tasks.
7//! `SqliteConnection` is a thin newtype that holds:
8//!
9//! - `conn: limbo::Connection` — the Limbo connection handle.
10//! - `txn_lock: Arc<tokio::sync::Mutex<()>>` — a guard that prevents two async tasks
11//! from issuing `BEGIN` concurrently on the same logical connection. SQLite does
12//! not support nested transactions, so only one task at a time may hold a
13//! transaction.
14//! - `path: String` — the path supplied to [`Builder::new_local`], retained for
15//! diagnostics.
16//!
17//! # Affected-row count
18//!
19//! After each DML statement we call `conn.changes()` to read the row count that
20//! was committed by the most-recent write transaction. DDL statements and
21//! `BEGIN`/`COMMIT`/`ROLLBACK` leave the counter at 0, which is the correct
22//! contract per OxiSQL and `sqlite3_changes()` semantics.
23//!
24//! # Parameter binding
25//!
26//! OxiSQL passes `$1`, `$2`, … positional parameters. SQLite / Limbo expects
27//! `?` placeholders. `types::rewrite_params` performs a quote-aware
28//! translation before each statement is prepared.
29//!
30//! # Schema introspection
31//!
32//! [`Connection::tables`] queries `sqlite_master`.
33//! [`Connection::columns`] uses `PRAGMA table_info`.
34//! [`Connection::indexes`] parses `sqlite_master` DDL (PRAGMA index_list/index_info are not
35//! yet implemented in Limbo 0.0.22).
36//! [`Connection::foreign_keys`] uses `PRAGMA foreign_key_list` — the engine now
37//! surfaces FK metadata from its in-memory schema.
38//!
39//! # Transactions
40//!
41//! [`Connection::transaction`] issues `BEGIN` and returns a [`SqliteTransaction`]
42//! that wraps the same `limbo::Connection`. The transaction holds a guard on
43//! `txn_lock` so that no other task can start a concurrent `BEGIN`.
44//! Dropping `SqliteTransaction` without calling `commit` or `rollback` will
45//! execute `ROLLBACK` (best-effort, via `Drop`).
46//!
47//! # Prepared-statement cache
48//!
49//! All DML and DDL statements pass through an LRU cache keyed by the
50//! **rewritten SQL** (after `$N`→`?` translation). The cache holds up to
51//! `STMT_CACHE_CAPACITY` (128) compiled `limbo::Statement` entries per connection
52//! (shared across clones of the same connection via `Arc<StdMutex<…>>`).
53//!
54//! On a cache hit the existing `limbo::Statement` is taken out of the cache,
55//! executed via `Statement::execute()` (which calls `reset()` before binding),
56//! and returned to the cache after execution. `Statement::reset()` now also
57//! zeroes `Program::n_change` (fixed in oxisqlite-core), so cached statement
58//! reuse produces correct per-execution change counts.
59//!
60//! # ROLLBACK
61//!
62//! `SqliteTransaction::rollback()` executes the SQL string `"ROLLBACK"` against
63//! the engine, exactly mirroring how `commit()` executes `"COMMIT"`. The engine
64//! emits an `AutoCommit { auto_commit: true, rollback: true }` VDBE instruction
65//! that discards all pending changes. The `Drop` impl also fires a best-effort
66//! ROLLBACK when the transaction is dropped without an explicit `commit()` or
67//! `rollback()`.
68//!
69//! # Prepared-statement reuse (via SqlitePrepared)
70//!
71//! Limbo's `Statement` is consumed after a single `execute`/`query` cycle.
72//! Our [`PreparedStatement`] wrapper therefore re-prepares on every call. The
73//! API contract (parse-once, bind-many) is satisfied at the OxiSQL trait level
74//! even though Limbo does not yet expose a stable compiled-statement cache.
75
76use std::num::NonZeroUsize;
77use std::sync::{Arc, Mutex as StdMutex};
78
79use async_trait::async_trait;
80use limbo::params::Params as LimboParams;
81use limbo::Builder;
82use tokio::sync::Mutex as TokioMutex;
83
84// ── statement-cache capacity ───────────────────────────────────────────────────
85
86/// Maximum number of compiled statements retained in the per-connection LRU
87/// cache. Statements are keyed by their rewritten SQL (`?`-placeholder form).
88const STMT_CACHE_CAPACITY: usize = 128;
89
90use oxisql_core::{
91 ColumnInfo, Connection, ForeignKeyInfo, IndexInfo, OxiSqlError, PreparedStatement, Row,
92 TableInfo, TableType, ToSqlValue, Transaction, Value,
93};
94
95use crate::error::SqliteCompatError;
96use crate::types::{limbo_to_core_typed, rewrite_params, split_statements};
97
98// ── helpers ───────────────────────────────────────────────────────────────────
99
100/// A per-connection LRU cache from rewritten SQL → compiled `limbo::Statement`.
101///
102/// Wrapped in `Arc<StdMutex<…>>` so it can be cheaply shared when the
103/// `SqliteConnection` is cloned. The std `Mutex` is deliberately chosen over
104/// `tokio::sync::Mutex`: the critical section is very short (single hash-lookup
105/// or insertion) and never held across an `.await` point.
106type StmtCache = Arc<StdMutex<lru::LruCache<String, limbo::Statement>>>;
107
108/// Construct a new, empty [`StmtCache`] with [`STMT_CACHE_CAPACITY`] slots.
109fn new_stmt_cache() -> StmtCache {
110 // SAFETY: STMT_CACHE_CAPACITY is a positive compile-time constant (128).
111 // `NonZeroUsize::new` returns `None` only for 0, which this is not.
112 let cap = NonZeroUsize::new(STMT_CACHE_CAPACITY).unwrap_or(NonZeroUsize::MIN);
113 Arc::new(StdMutex::new(lru::LruCache::new(cap)))
114}
115
116/// Execute a SQL statement that has already been rewritten to `?` placeholders.
117///
118/// All statements (DML and DDL) pass through the statement cache uniformly.
119/// On a cache miss the statement is compiled via `conn.prepare()`, executed,
120/// and stored for future reuse. On a cache hit the existing `limbo::Statement`
121/// is retrieved, executed via `stmt.execute()` (which calls `reset()` before
122/// binding, zeroing `n_change` so reuse produces correct per-execution change
123/// counts), and returned to the cache.
124///
125/// If the cached statement was compiled before a schema change (DDL, ALTER,
126/// CREATE INDEX, etc.), the engine's `op_transaction` cookie check fires on
127/// the first `step()` and returns `SchemaChanged`. This function catches that
128/// error, discards the stale compiled program, re-prepares against the
129/// refreshed schema, and retries exactly once. This transparent re-prepare
130/// replaces the old `is_ddl` keyword-prefix heuristic that failed on
131/// comment-prefixed DDL and left DML statements stale after schema changes.
132///
133/// The affected-row count is read from `conn.changes()` after execution,
134/// which reflects the count committed by the most recent write transaction on
135/// this connection. DDL and `BEGIN`/`COMMIT`/`ROLLBACK` return 0, which is the
136/// correct value per the OxiSQL contract.
137///
138/// When no `cache` is provided (e.g., in unit tests that bypass the cache) the
139/// function falls back to `conn.execute()` followed by `conn.changes()`.
140async fn exec_rewritten(
141 conn: &limbo::Connection,
142 sql: &str,
143 limbo_params: Vec<limbo::Value>,
144 cache: Option<&StmtCache>,
145) -> Result<u64, SqliteCompatError> {
146 match cache {
147 Some(c) => {
148 // Clone before consuming so we can rebuild the parameter list for a
149 // re-prepare-and-retry if the engine signals SchemaChanged.
150 let retry_params = limbo_params.clone();
151 let lp = if limbo_params.is_empty() {
152 LimboParams::None
153 } else {
154 LimboParams::Positional(limbo_params)
155 };
156
157 // Take the compiled statement out of the cache (if present).
158 // The lock is held only for this short lookup; never across `.await`.
159 let cached = {
160 let mut guard = c.lock().map_err(|e| {
161 SqliteCompatError::Other(format!("stmt_cache lock poisoned: {e}"))
162 })?;
163 guard.pop(sql)
164 };
165
166 let mut stmt = match cached {
167 Some(s) => s,
168 None => conn.prepare(sql).await.map_err(SqliteCompatError::from)?,
169 };
170
171 match stmt.execute(lp).await {
172 Ok(_) => {
173 // Execution succeeded — return the statement to the cache.
174 c.lock()
175 .map_err(|e| {
176 SqliteCompatError::Other(format!("stmt_cache lock poisoned: {e}"))
177 })?
178 .put(sql.to_owned(), stmt);
179 }
180 Err(e) if e.is_schema_changed() => {
181 // The schema changed after this statement was compiled. Drop
182 // the stale program, re-compile against the refreshed schema,
183 // and retry exactly once.
184 drop(stmt);
185 let retry_lp = if retry_params.is_empty() {
186 LimboParams::None
187 } else {
188 LimboParams::Positional(retry_params)
189 };
190 let mut fresh = conn.prepare(sql).await.map_err(SqliteCompatError::from)?;
191 fresh
192 .execute(retry_lp)
193 .await
194 .map_err(SqliteCompatError::from)?;
195 c.lock()
196 .map_err(|e| {
197 SqliteCompatError::Other(format!("stmt_cache lock poisoned: {e}"))
198 })?
199 .put(sql.to_owned(), fresh);
200 }
201 Err(e) => return Err(SqliteCompatError::from(e)),
202 }
203
204 let n = conn
205 .changes()
206 .map_err(|e| SqliteCompatError::Other(format!("changes() failed: {e}")))?;
207 Ok(n.max(0) as u64)
208 }
209 None => {
210 // ── no-cache path (uncommon; bypasses the cache entirely) ──────────
211 let lp = if limbo_params.is_empty() {
212 LimboParams::None
213 } else {
214 LimboParams::Positional(limbo_params)
215 };
216 conn.execute(sql, lp)
217 .await
218 .map_err(SqliteCompatError::from)?;
219 let n = conn
220 .changes()
221 .map_err(|e| SqliteCompatError::Other(format!("changes() failed: {e}")))?;
222 Ok(n.max(0) as u64)
223 }
224 }
225}
226
227/// Execute a query that has already been rewritten to `?` placeholders and
228/// collect all result rows.
229///
230/// Column declared types (e.g. `"DATE"`, `"TIMESTAMP"`, `"UUID"`) are
231/// collected from the prepared statement and forwarded to [`limbo_to_core_typed`]
232/// so that richer [`Value`] variants are produced when appropriate.
233async fn query_rewritten(
234 conn: &limbo::Connection,
235 sql: &str,
236 limbo_params: Vec<limbo::Value>,
237) -> Result<Vec<Row>, SqliteCompatError> {
238 let lp = if limbo_params.is_empty() {
239 LimboParams::None
240 } else {
241 LimboParams::Positional(limbo_params)
242 };
243
244 let mut stmt = conn.prepare(sql).await.map_err(SqliteCompatError::from)?;
245
246 // Collect column names and declared types together.
247 let col_info: Vec<(String, Option<String>)> = stmt
248 .columns()
249 .iter()
250 .map(|c| (c.name().to_owned(), c.decl_type().map(str::to_owned)))
251 .collect();
252
253 let col_names: Vec<String> = col_info.iter().map(|(name, _)| name.clone()).collect();
254
255 let mut rows_iter = stmt.query(lp).await.map_err(SqliteCompatError::from)?;
256
257 let mut rows: Vec<Row> = Vec::new();
258 while let Some(limbo_row) = rows_iter.next().await.map_err(SqliteCompatError::from)? {
259 let mut values: Vec<Value> = Vec::with_capacity(col_info.len());
260 for idx in 0..limbo_row.column_count() {
261 let raw = limbo_row.get_value(idx).map_err(SqliteCompatError::from)?;
262 let decl = col_info.get(idx).and_then(|(_, dt)| dt.as_deref());
263 values.push(limbo_to_core_typed(raw, decl)?);
264 }
265 rows.push(Row::new(col_names.clone(), values));
266 }
267 Ok(rows)
268}
269
270// ── SqliteConnection ──────────────────────────────────────────────────────────
271
272/// A Limbo-backed SQLite connection implementing [`Connection`].
273///
274/// Create via [`SqliteConnection::open`] (file path) or
275/// [`SqliteConnection::open_memory`] (`:memory:`).
276///
277/// # Statement cache
278///
279/// Each `SqliteConnection` maintains an LRU cache of compiled `limbo::Statement`
280/// objects (capacity: `STMT_CACHE_CAPACITY` = 128). The cache is shared across
281/// clones of the same connection (the clones share the underlying
282/// `limbo::Connection`) and is updated on every DML/DDL execution. Cache hits
283/// save the per-statement parse-and-compile round-trip inside Limbo.
284#[derive(Clone)]
285pub struct SqliteConnection {
286 conn: limbo::Connection,
287 txn_lock: Arc<TokioMutex<()>>,
288 stmt_cache: StmtCache,
289 path: String,
290}
291
292impl std::fmt::Debug for SqliteConnection {
293 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
294 let cache_len = self.stmt_cache.lock().map(|g| g.len()).unwrap_or(0);
295 f.debug_struct("SqliteConnection")
296 .field("path", &self.path)
297 .field("stmt_cache_len", &cache_len)
298 .finish_non_exhaustive()
299 }
300}
301
302impl SqliteConnection {
303 /// Open a Limbo database at the given file path.
304 ///
305 /// Pass `":memory:"` for an in-memory database, or use
306 /// [`open_memory`][Self::open_memory] for clarity.
307 ///
308 /// # Errors
309 ///
310 /// Returns [`OxiSqlError`] if the file cannot be opened or created.
311 pub async fn open(path: &str) -> Result<Self, OxiSqlError> {
312 let db = Builder::new_local(path)
313 .build()
314 .await
315 .map_err(|e| OxiSqlError::Other(format!("limbo open error: {e}")))?;
316 let conn = db
317 .connect()
318 .map_err(|e| OxiSqlError::Other(format!("limbo connect error: {e}")))?;
319 Ok(Self {
320 conn,
321 txn_lock: Arc::new(TokioMutex::new(())),
322 stmt_cache: new_stmt_cache(),
323 path: path.to_owned(),
324 })
325 }
326
327 /// Open a fresh in-memory Limbo database.
328 ///
329 /// # Errors
330 ///
331 /// Returns [`OxiSqlError`] if the engine cannot be initialised.
332 pub async fn open_memory() -> Result<Self, OxiSqlError> {
333 Self::open(":memory:").await
334 }
335
336 /// Open a Limbo-backed connection from an in-memory SQLite database image.
337 ///
338 /// The `bytes` are copied into an in-memory page store; no temporary file
339 /// is ever created, so this works on WASI, in the browser, and on
340 /// read-only filesystems. Mirrors SQLite's `sqlite3_deserialize()` /
341 /// rusqlite's `Connection::deserialize`. See
342 /// [`limbo::Database::open_from_bytes`].
343 ///
344 /// # Example
345 ///
346 /// ```rust,no_run
347 /// # async fn run() -> Result<(), oxisql_core::OxiSqlError> {
348 /// use oxisql_core::Connection;
349 /// use oxisql_sqlite_compat::SqliteConnection;
350 ///
351 /// // `image` is a complete SQLite database file loaded into memory,
352 /// // e.g. `include_bytes!("../data/app.db")`.
353 /// let image: &[u8] = get_database_image();
354 /// let conn = SqliteConnection::open_from_bytes(image).await?;
355 /// let rows = conn.query("SELECT count(*) FROM sqlite_master", &[]).await?;
356 /// # let _ = rows;
357 /// # Ok(())
358 /// # }
359 /// # fn get_database_image() -> &'static [u8] { &[] }
360 /// ```
361 ///
362 /// # Errors
363 ///
364 /// Returns [`OxiSqlError`] if `bytes` is not a valid SQLite database image
365 /// (too short, wrong magic header, or an invalid page size). Never panics
366 /// on malformed input.
367 pub async fn open_from_bytes(bytes: &[u8]) -> Result<Self, OxiSqlError> {
368 let db = limbo::Database::open_from_bytes(bytes)
369 .map_err(|e| OxiSqlError::Other(format!("limbo open_from_bytes error: {e}")))?;
370 let conn = db
371 .connect()
372 .map_err(|e| OxiSqlError::Other(format!("limbo connect error: {e}")))?;
373 Ok(Self {
374 conn,
375 txn_lock: Arc::new(TokioMutex::new(())),
376 stmt_cache: new_stmt_cache(),
377 path: "<memory:bytes>".to_owned(),
378 })
379 }
380
381 /// Return the path this connection was opened with.
382 pub fn path(&self) -> &str {
383 &self.path
384 }
385}
386
387// ── Connection impl ───────────────────────────────────────────────────────────
388
389#[async_trait]
390impl Connection for SqliteConnection {
391 async fn execute(&self, sql: &str, params: &[&dyn ToSqlValue]) -> Result<u64, OxiSqlError> {
392 let (rewritten, limbo_params) = rewrite_params(sql, params).map_err(OxiSqlError::from)?;
393 exec_rewritten(&self.conn, &rewritten, limbo_params, Some(&self.stmt_cache))
394 .await
395 .map_err(OxiSqlError::from)
396 }
397
398 async fn query(&self, sql: &str, params: &[&dyn ToSqlValue]) -> Result<Vec<Row>, OxiSqlError> {
399 let (rewritten, limbo_params) = rewrite_params(sql, params).map_err(OxiSqlError::from)?;
400 query_rewritten(&self.conn, &rewritten, limbo_params)
401 .await
402 .map_err(OxiSqlError::from)
403 }
404
405 async fn transaction(&self) -> Result<Box<dyn Transaction + '_>, OxiSqlError> {
406 // Acquire the exclusive transaction lock before issuing BEGIN.
407 // This prevents a second task from starting a concurrent transaction
408 // on the same SqliteConnection clone.
409 let guard = self.txn_lock.lock().await;
410 self.conn
411 .execute("BEGIN", LimboParams::None)
412 .await
413 .map_err(|e| OxiSqlError::Other(format!("BEGIN failed: {e}")))?;
414 Ok(Box::new(SqliteTransaction {
415 conn: self.conn.clone(),
416 // Share the connection-level stmt_cache so that DML executed inside
417 // a transaction also benefits from cached compiled statements.
418 stmt_cache: Arc::clone(&self.stmt_cache),
419 // Transfer ownership of the mutex guard into the transaction.
420 // The guard is released when SqliteTransaction is dropped.
421 _guard: guard,
422 done: false,
423 }))
424 }
425
426 async fn execute_batch(&self, sql: &str) -> Result<u64, OxiSqlError> {
427 // Token-aware split: honours `;` inside string literals, quoted
428 // identifiers, block comments, and line comments.
429 let stmts = split_statements(sql);
430 let mut total = 0u64;
431 for stmt in stmts {
432 total += self.execute(stmt, &[]).await?;
433 }
434 Ok(total)
435 }
436
437 async fn ping(&self) -> Result<(), OxiSqlError> {
438 self.query("SELECT 1", &[]).await?;
439 Ok(())
440 }
441
442 async fn prepare(&self, sql: &str) -> Result<Box<dyn PreparedStatement + '_>, OxiSqlError> {
443 Ok(Box::new(SqlitePrepared {
444 conn: &self.conn,
445 stmt_cache: Arc::clone(&self.stmt_cache),
446 sql: sql.to_owned(),
447 }))
448 }
449
450 // ── Schema introspection ──────────────────────────────────────────────────
451
452 async fn tables(&self) -> Result<Vec<TableInfo>, OxiSqlError> {
453 let rows = self
454 .query(
455 "SELECT name, type FROM sqlite_master \
456 WHERE type IN ('table','view') AND name NOT LIKE 'sqlite_%' \
457 ORDER BY name",
458 &[],
459 )
460 .await?;
461
462 let infos = rows
463 .into_iter()
464 .map(|row| {
465 let name = row
466 .get_by_index(0)
467 .and_then(|v| {
468 if let Value::Text(s) = v {
469 Some(s.clone())
470 } else {
471 None
472 }
473 })
474 .unwrap_or_default();
475 let ttype_str = row
476 .get_by_index(1)
477 .and_then(|v| {
478 if let Value::Text(s) = v {
479 Some(s.as_str())
480 } else {
481 None
482 }
483 })
484 .unwrap_or("table");
485 let table_type = match ttype_str {
486 "view" => TableType::View,
487 _ => TableType::Base,
488 };
489 TableInfo {
490 name,
491 schema: None,
492 table_type,
493 }
494 })
495 .collect();
496 Ok(infos)
497 }
498
499 async fn columns(&self, table: &str) -> Result<Vec<ColumnInfo>, OxiSqlError> {
500 // PRAGMA table_info returns: cid, name, type, notnull, dflt_value, pk
501 let sql = format!("PRAGMA table_info(\"{table}\")");
502 let rows = self.query(&sql, &[]).await?;
503
504 let infos = rows
505 .into_iter()
506 .map(|row| {
507 // Helper: get column by index as string or empty string.
508 let text_at = |r: &Row, idx: usize| -> String {
509 r.get_by_index(idx)
510 .and_then(|v| match v {
511 Value::Text(s) => Some(s.clone()),
512 Value::I64(n) => Some(n.to_string()),
513 Value::Null => Some(String::new()),
514 _ => None,
515 })
516 .unwrap_or_default()
517 };
518 let i64_at = |r: &Row, idx: usize| -> i64 {
519 r.get_by_index(idx)
520 .and_then(|v| {
521 if let Value::I64(n) = v {
522 Some(*n)
523 } else {
524 None
525 }
526 })
527 .unwrap_or(0)
528 };
529
530 let ordinal = i64_at(&row, 0) as u32 + 1; // cid is 0-based
531 let name = text_at(&row, 1);
532 let data_type = text_at(&row, 2);
533 let notnull = i64_at(&row, 3) != 0;
534 let default_val = row.get_by_index(4).and_then(|v| match v {
535 Value::Text(s) => Some(s.clone()),
536 Value::Null => None,
537 other => Some(format!("{other:?}")),
538 });
539
540 ColumnInfo {
541 name,
542 ordinal_position: ordinal,
543 data_type,
544 nullable: !notnull,
545 default: default_val,
546 max_length: None,
547 numeric_precision: None,
548 numeric_scale: None,
549 }
550 })
551 .collect();
552 Ok(infos)
553 }
554
555 async fn indexes(&self, table: &str) -> Result<Vec<IndexInfo>, OxiSqlError> {
556 // PRAGMA index_list and PRAGMA index_info are not yet implemented in limbo 0.0.22.
557 // Fall back to sqlite_master for index names and uniqueness, then parse
558 // the index SQL to extract column names. This is best-effort: multi-column
559 // indexes and expression indexes may not parse perfectly.
560 let sql = "SELECT name, sql FROM sqlite_master \
561 WHERE type='index' AND tbl_name=$1 AND name NOT LIKE 'sqlite_%'";
562 let rows = self.query(sql, &[&table]).await?;
563
564 let mut infos: Vec<IndexInfo> = Vec::new();
565 for row in rows {
566 let name = row
567 .get_by_index(0)
568 .and_then(|v| {
569 if let Value::Text(s) = v {
570 Some(s.clone())
571 } else {
572 None
573 }
574 })
575 .unwrap_or_default();
576 let idx_sql = row
577 .get_by_index(1)
578 .and_then(|v| {
579 if let Value::Text(s) = v {
580 Some(s.clone())
581 } else {
582 None
583 }
584 })
585 .unwrap_or_default();
586
587 // Detect UNIQUE from the CREATE INDEX / CREATE UNIQUE INDEX statement.
588 let upper = idx_sql.to_ascii_uppercase();
589 let unique = upper.contains("UNIQUE");
590
591 // Extract column list between the last `(` and `)`.
592 let columns: Vec<String> =
593 if let (Some(open), Some(close)) = (idx_sql.rfind('('), idx_sql.rfind(')')) {
594 idx_sql[open + 1..close]
595 .split(',')
596 .map(|c| c.trim().to_string())
597 .filter(|c| !c.is_empty())
598 .collect()
599 } else {
600 vec![]
601 };
602
603 infos.push(IndexInfo {
604 name,
605 columns,
606 unique,
607 primary: false,
608 });
609 }
610 Ok(infos)
611 }
612
613 async fn foreign_keys(&self, table: &str) -> Result<Vec<ForeignKeyInfo>, OxiSqlError> {
614 // Use PRAGMA foreign_key_list — the engine now surfaces FK metadata
615 // directly from the in-memory schema, avoiding brittle DDL text parsing.
616 let escaped = table.replace('"', "\"\"");
617 let sql = format!("PRAGMA foreign_key_list(\"{}\")", escaped);
618 let rows = query_rewritten(&self.conn, &sql, vec![])
619 .await
620 .map_err(OxiSqlError::from)?;
621
622 // PRAGMA foreign_key_list columns (by index):
623 // 0: id INTEGER — FK index within the table
624 // 1: seq INTEGER — column position within a composite FK
625 // 2: table TEXT — parent table name
626 // 3: from TEXT — child column name
627 // 4: to TEXT/NULL — parent column name (NULL = implicit PK ref)
628 // 5: on_update TEXT
629 // 6: on_delete TEXT
630 // 7: match TEXT
631 let mut infos: Vec<ForeignKeyInfo> = Vec::with_capacity(rows.len());
632 for row in &rows {
633 let id = match row.get_by_index(0) {
634 Some(Value::I64(v)) => *v,
635 _ => 0,
636 };
637 let from_col = match row.get_by_index(3) {
638 Some(Value::Text(s)) => s.clone(),
639 _ => continue,
640 };
641 let foreign_table = match row.get_by_index(2) {
642 Some(Value::Text(s)) => s.clone(),
643 _ => continue,
644 };
645 let foreign_column = match row.get_by_index(4) {
646 Some(Value::Text(s)) => s.clone(),
647 _ => String::new(),
648 };
649 let on_update = match row.get_by_index(5) {
650 Some(Value::Text(s)) => Some(s.clone()),
651 _ => None,
652 };
653 let on_delete = match row.get_by_index(6) {
654 Some(Value::Text(s)) => Some(s.clone()),
655 _ => None,
656 };
657 let constraint_name = format!("fk_{table}_{id}");
658 infos.push(ForeignKeyInfo {
659 constraint_name,
660 column: from_col,
661 foreign_table,
662 foreign_column,
663 on_update,
664 on_delete,
665 });
666 }
667 Ok(infos)
668 }
669}
670
671// ── SqliteTransaction ─────────────────────────────────────────────────────────
672
673/// A SQLite transaction backed by raw `BEGIN`/`COMMIT`/`ROLLBACK` statements.
674///
675/// Holds a guard on the connection-level transaction mutex so that no other
676/// async task can start a concurrent `BEGIN` on the same `SqliteConnection`.
677/// When dropped without an explicit `commit` or `rollback`, the transaction
678/// attempts a best-effort `ROLLBACK` via a background task.
679pub struct SqliteTransaction<'a> {
680 conn: limbo::Connection,
681 stmt_cache: StmtCache,
682 _guard: tokio::sync::MutexGuard<'a, ()>,
683 done: bool,
684}
685
686impl<'a> Drop for SqliteTransaction<'a> {
687 fn drop(&mut self) {
688 if !self.done {
689 // Best-effort rollback on implicit drop. We cannot `.await` inside
690 // `drop`, so we spawn a fire-and-forget task. The mutex guard is
691 // released when `SqliteTransaction` is fully dropped (after this
692 // function body returns).
693 let conn = self.conn.clone();
694 tokio::spawn(async move {
695 if let Err(e) = conn.execute("ROLLBACK", LimboParams::None).await {
696 log::warn!("SqliteTransaction drop: ROLLBACK failed: {e}");
697 }
698 });
699 }
700 }
701}
702
703#[async_trait]
704impl<'a> Transaction for SqliteTransaction<'a> {
705 async fn execute(&mut self, sql: &str, params: &[&dyn ToSqlValue]) -> Result<u64, OxiSqlError> {
706 let (rewritten, limbo_params) = rewrite_params(sql, params).map_err(OxiSqlError::from)?;
707 exec_rewritten(&self.conn, &rewritten, limbo_params, Some(&self.stmt_cache))
708 .await
709 .map_err(OxiSqlError::from)
710 }
711
712 async fn query(
713 &mut self,
714 sql: &str,
715 params: &[&dyn ToSqlValue],
716 ) -> Result<Vec<Row>, OxiSqlError> {
717 let (rewritten, limbo_params) = rewrite_params(sql, params).map_err(OxiSqlError::from)?;
718 query_rewritten(&self.conn, &rewritten, limbo_params)
719 .await
720 .map_err(OxiSqlError::from)
721 }
722
723 async fn commit(mut self: Box<Self>) -> Result<(), OxiSqlError> {
724 self.done = true;
725 self.conn
726 .execute("COMMIT", LimboParams::None)
727 .await
728 .map_err(|e| OxiSqlError::Other(format!("COMMIT failed: {e}")))?;
729 Ok(())
730 }
731
732 async fn rollback(mut self: Box<Self>) -> Result<(), OxiSqlError> {
733 // Mark done so that Drop does not attempt a second ROLLBACK.
734 self.done = true;
735 self.conn
736 .execute("ROLLBACK", LimboParams::None)
737 .await
738 .map_err(|e| OxiSqlError::Other(format!("ROLLBACK failed: {e}")))?;
739 Ok(())
740 }
741}
742
743// ── SqlitePrepared ────────────────────────────────────────────────────────────
744
745/// A prepared statement backed by the connection-level LRU cache.
746///
747/// On each `execute()` call the cached `limbo::Statement` is retrieved (or
748/// compiled fresh on a miss), executed, and returned to the cache. Because
749/// `Statement::reset()` now zeroes `n_change`, every execution sees a correct
750/// change count without re-parsing the SQL.
751pub struct SqlitePrepared<'a> {
752 conn: &'a limbo::Connection,
753 stmt_cache: StmtCache,
754 sql: String,
755}
756
757#[async_trait]
758impl<'a> PreparedStatement for SqlitePrepared<'a> {
759 async fn execute(&mut self, params: &[&dyn ToSqlValue]) -> Result<u64, OxiSqlError> {
760 let (rewritten, limbo_params) =
761 rewrite_params(&self.sql, params).map_err(OxiSqlError::from)?;
762 exec_rewritten(self.conn, &rewritten, limbo_params, Some(&self.stmt_cache))
763 .await
764 .map_err(OxiSqlError::from)
765 }
766
767 async fn query(&mut self, params: &[&dyn ToSqlValue]) -> Result<Vec<Row>, OxiSqlError> {
768 let (rewritten, limbo_params) =
769 rewrite_params(&self.sql, params).map_err(OxiSqlError::from)?;
770 query_rewritten(self.conn, &rewritten, limbo_params)
771 .await
772 .map_err(OxiSqlError::from)
773 }
774
775 fn sql(&self) -> &str {
776 &self.sql
777 }
778}