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