syncular_runtime/storage/
sqlite_query.rs

1use crate::app_schema::{default_app_schema, AppSchema};
2use crate::error::{ErrorKind, Result, SyncularError};
3use crate::store::SQLITE_BUSY_TIMEOUT_MS;
4use libsqlite3_sys as sqlite;
5use serde::{Deserialize, Serialize};
6use serde_json::{Map, Number, Value};
7use std::collections::BTreeSet;
8use std::collections::{BTreeMap, VecDeque};
9use std::ffi::{CStr, CString};
10use std::os::raw::{c_char, c_int, c_void};
11use std::ptr;
12use std::slice;
13
14#[derive(Debug, Clone, Deserialize)]
15#[serde(rename_all = "camelCase")]
16pub struct ReadonlySqlQueryRequest {
17    pub sql: String,
18    #[serde(default)]
19    pub params: Vec<Value>,
20    #[serde(default)]
21    pub tables: Vec<String>,
22}
23
24#[derive(Debug, Clone, Serialize)]
25#[serde(rename_all = "camelCase")]
26pub struct ReadonlySqlQueryResult {
27    pub rows: Vec<Value>,
28}
29
30struct SqliteDb {
31    raw: *mut sqlite::sqlite3,
32}
33
34struct SqliteStatement {
35    raw: *mut sqlite::sqlite3_stmt,
36}
37
38unsafe impl Send for SqliteDb {}
39unsafe impl Send for SqliteStatement {}
40
41pub struct ReadonlySqlQueryExecutor {
42    db: SqliteDb,
43    app_schema: AppSchema,
44    schema_version: i32,
45    capacity: usize,
46    statements: BTreeMap<String, SqliteStatement>,
47    statement_order: VecDeque<String>,
48}
49
50struct AuthorizerContext {
51    allowed_tables: BTreeSet<String>,
52    denied: Option<String>,
53}
54
55impl Drop for SqliteDb {
56    fn drop(&mut self) {
57        if !self.raw.is_null() {
58            unsafe {
59                sqlite::sqlite3_close(self.raw);
60            }
61        }
62    }
63}
64
65impl Drop for SqliteStatement {
66    fn drop(&mut self) {
67        if !self.raw.is_null() {
68            unsafe {
69                sqlite::sqlite3_finalize(self.raw);
70            }
71        }
72    }
73}
74
75pub fn execute_readonly_query_json(db_path: &str, request_json: &str) -> Result<String> {
76    let request: ReadonlySqlQueryRequest = serde_json::from_str(request_json)?;
77    let result = execute_readonly_query_with_schema(db_path, request, default_app_schema())?;
78    Ok(serde_json::to_string(&result)?)
79}
80
81pub fn execute_readonly_query_json_with_schema(
82    db_path: &str,
83    request_json: &str,
84    app_schema: AppSchema,
85) -> Result<String> {
86    let request: ReadonlySqlQueryRequest = serde_json::from_str(request_json)?;
87    let result = execute_readonly_query_with_schema(db_path, request, app_schema)?;
88    Ok(serde_json::to_string(&result)?)
89}
90
91pub fn execute_readonly_query(
92    db_path: &str,
93    request: ReadonlySqlQueryRequest,
94) -> Result<ReadonlySqlQueryResult> {
95    execute_readonly_query_with_schema(db_path, request, default_app_schema())
96}
97
98pub fn execute_readonly_query_with_schema(
99    db_path: &str,
100    request: ReadonlySqlQueryRequest,
101    app_schema: AppSchema,
102) -> Result<ReadonlySqlQueryResult> {
103    if request.sql.trim().is_empty() {
104        return Err(SyncularError::config("query SQL must not be empty"));
105    }
106
107    let db = open_db(db_path)?;
108    let allowed_tables = validate_tables(&request.tables, app_schema)?;
109    let stmt = prepare_authorized_statement(&db, &request.sql, allowed_tables)?;
110
111    bind_params(&db, &stmt, &request.params)?;
112    let rows = read_rows(&db, &stmt)?;
113
114    Ok(ReadonlySqlQueryResult { rows })
115}
116
117impl ReadonlySqlQueryExecutor {
118    pub fn open(db_path: &str, app_schema: AppSchema, capacity: usize) -> Result<Self> {
119        Ok(Self {
120            db: open_db(db_path)?,
121            app_schema,
122            schema_version: app_schema.current_schema_version(),
123            capacity: capacity.max(1),
124            statements: BTreeMap::new(),
125            statement_order: VecDeque::new(),
126        })
127    }
128
129    pub fn execute_json(&mut self, request_json: &str) -> Result<String> {
130        let request: ReadonlySqlQueryRequest = serde_json::from_str(request_json)?;
131        let result = self.execute(request)?;
132        Ok(serde_json::to_string(&result)?)
133    }
134
135    pub fn execute(&mut self, request: ReadonlySqlQueryRequest) -> Result<ReadonlySqlQueryResult> {
136        if request.sql.trim().is_empty() {
137            return Err(SyncularError::config("query SQL must not be empty"));
138        }
139
140        let allowed_tables = validate_tables(&request.tables, self.app_schema)?;
141        let cache_key =
142            readonly_query_cache_key(&request.sql, &allowed_tables, self.schema_version);
143        if !self.statements.contains_key(&cache_key) {
144            let statement = prepare_authorized_statement(&self.db, &request.sql, allowed_tables)?;
145            self.insert_statement(cache_key.clone(), statement);
146        }
147
148        let stmt = self
149            .statements
150            .get_mut(&cache_key)
151            .ok_or_else(|| SyncularError::message(ErrorKind::Internal, "query cache miss"))?;
152        let result =
153            bind_params(&self.db, stmt, &request.params).and_then(|_| read_rows(&self.db, stmt));
154        reset_statement(&self.db, stmt)?;
155        Ok(ReadonlySqlQueryResult { rows: result? })
156    }
157
158    fn insert_statement(&mut self, cache_key: String, statement: SqliteStatement) {
159        if self.statements.len() >= self.capacity {
160            if let Some(oldest) = self.statement_order.pop_front() {
161                self.statements.remove(&oldest);
162            }
163        }
164        self.statement_order.push_back(cache_key.clone());
165        self.statements.insert(cache_key, statement);
166    }
167}
168
169fn validate_tables(tables: &[String], app_schema: AppSchema) -> Result<BTreeSet<String>> {
170    let mut allowed = BTreeSet::new();
171    for table in tables {
172        let table = table.trim();
173        if table.is_empty() {
174            return Err(SyncularError::config("query table dependency is empty"));
175        }
176        if app_schema.table_metadata(table).is_none() {
177            return Err(SyncularError::config(format!(
178                "queryJson can only read generated app tables; unknown table: {table}"
179            )));
180        }
181        allowed.insert(table.to_string());
182    }
183    Ok(allowed)
184}
185
186fn readonly_query_cache_key(
187    sql: &str,
188    allowed_tables: &BTreeSet<String>,
189    schema_version: i32,
190) -> String {
191    format!(
192        "{}\u{1f}{}\u{1f}{}",
193        schema_version,
194        sql,
195        allowed_tables.iter().cloned().collect::<Vec<_>>().join(",")
196    )
197}
198
199fn open_db(db_path: &str) -> Result<SqliteDb> {
200    let path = CString::new(db_path)
201        .map_err(|_| SyncularError::config("database path contains interior NUL byte"))?;
202    let mut raw = ptr::null_mut();
203    let rc = unsafe {
204        sqlite::sqlite3_open_v2(
205            path.as_ptr(),
206            &mut raw,
207            sqlite::SQLITE_OPEN_READONLY,
208            ptr::null(),
209        )
210    };
211    if rc != sqlite::SQLITE_OK {
212        let message = sqlite_message(raw, "open read-only query connection");
213        if !raw.is_null() {
214            unsafe {
215                sqlite::sqlite3_close(raw);
216            }
217        }
218        return Err(message);
219    }
220
221    unsafe {
222        sqlite::sqlite3_busy_timeout(raw, SQLITE_BUSY_TIMEOUT_MS);
223    }
224
225    Ok(SqliteDb { raw })
226}
227
228fn install_authorizer(db: &SqliteDb, context: &mut AuthorizerContext) -> Result<()> {
229    let rc = unsafe {
230        sqlite::sqlite3_set_authorizer(
231            db.raw,
232            Some(readonly_authorizer),
233            context as *mut AuthorizerContext as *mut c_void,
234        )
235    };
236    if rc != sqlite::SQLITE_OK {
237        return Err(sqlite_message(db.raw, "install query authorizer"));
238    }
239    Ok(())
240}
241
242fn clear_authorizer(db: &SqliteDb) {
243    unsafe {
244        sqlite::sqlite3_set_authorizer(db.raw, None, ptr::null_mut());
245    }
246}
247
248fn prepare_authorized_statement(
249    db: &SqliteDb,
250    sql: &str,
251    allowed_tables: BTreeSet<String>,
252) -> Result<SqliteStatement> {
253    let mut authorizer = AuthorizerContext {
254        allowed_tables,
255        denied: None,
256    };
257    install_authorizer(db, &mut authorizer)?;
258    let result = prepare_single_statement(db, sql);
259    clear_authorizer(db);
260
261    match result {
262        Ok(stmt) => {
263            if unsafe { sqlite::sqlite3_stmt_readonly(stmt.raw) } == 0 {
264                return Err(SyncularError::config(
265                    "queryJson only accepts read-only SQL; use applyMutationJson for Syncular writes",
266                ));
267            }
268            if let Some(message) = authorizer.denied {
269                return Err(SyncularError::config(message));
270            }
271            Ok(stmt)
272        }
273        Err(error) => {
274            if let Some(message) = authorizer.denied {
275                Err(SyncularError::config(message))
276            } else {
277                Err(error)
278            }
279        }
280    }
281}
282
283fn prepare_single_statement(db: &SqliteDb, sql: &str) -> Result<SqliteStatement> {
284    let sql = CString::new(sql)
285        .map_err(|_| SyncularError::config("query SQL contains interior NUL byte"))?;
286    let mut raw = ptr::null_mut();
287    let mut tail = ptr::null();
288    let rc = unsafe { sqlite::sqlite3_prepare_v2(db.raw, sql.as_ptr(), -1, &mut raw, &mut tail) };
289    if rc != sqlite::SQLITE_OK {
290        return Err(sqlite_message(db.raw, "prepare read-only query"));
291    }
292    if raw.is_null() {
293        return Err(SyncularError::config(
294            "query SQL did not prepare a statement",
295        ));
296    }
297    if !tail.is_null() {
298        let tail = unsafe { CStr::from_ptr(tail) }
299            .to_string_lossy()
300            .trim()
301            .to_string();
302        if !tail.is_empty() {
303            unsafe {
304                sqlite::sqlite3_finalize(raw);
305            }
306            return Err(SyncularError::config(
307                "queryJson accepts exactly one SQL statement",
308            ));
309        }
310    }
311
312    Ok(SqliteStatement { raw })
313}
314
315fn reset_statement(db: &SqliteDb, stmt: &SqliteStatement) -> Result<()> {
316    let reset = unsafe { sqlite::sqlite3_reset(stmt.raw) };
317    let clear = unsafe { sqlite::sqlite3_clear_bindings(stmt.raw) };
318    if reset != sqlite::SQLITE_OK {
319        return Err(sqlite_message(db.raw, "reset read-only query"));
320    }
321    if clear != sqlite::SQLITE_OK {
322        return Err(sqlite_message(db.raw, "clear read-only query bindings"));
323    }
324    Ok(())
325}
326
327fn bind_params(db: &SqliteDb, stmt: &SqliteStatement, params: &[Value]) -> Result<()> {
328    let expected = unsafe { sqlite::sqlite3_bind_parameter_count(stmt.raw) };
329    if expected as usize != params.len() {
330        return Err(SyncularError::config(format!(
331            "query parameter count mismatch: SQL expects {expected}, request supplied {}",
332            params.len()
333        )));
334    }
335
336    for (index, value) in params.iter().enumerate() {
337        let parameter = (index + 1) as c_int;
338        let rc = match value {
339            Value::Null => unsafe { sqlite::sqlite3_bind_null(stmt.raw, parameter) },
340            Value::Bool(value) => unsafe {
341                sqlite::sqlite3_bind_int64(stmt.raw, parameter, i64::from(*value))
342            },
343            Value::Number(number) => bind_number(stmt.raw, parameter, number),
344            Value::String(value) => bind_text(stmt.raw, parameter, value)?,
345            Value::Array(_) | Value::Object(_) => {
346                return Err(SyncularError::config(
347                    "query params must be scalar JSON values",
348                ));
349            }
350        };
351        if rc != sqlite::SQLITE_OK {
352            return Err(sqlite_message(db.raw, "bind query parameter"));
353        }
354    }
355
356    Ok(())
357}
358
359fn bind_number(stmt: *mut sqlite::sqlite3_stmt, parameter: c_int, number: &Number) -> c_int {
360    if let Some(value) = number.as_i64() {
361        unsafe { sqlite::sqlite3_bind_int64(stmt, parameter, value) }
362    } else if let Some(value) = number.as_u64() {
363        if value <= i64::MAX as u64 {
364            unsafe { sqlite::sqlite3_bind_int64(stmt, parameter, value as i64) }
365        } else {
366            sqlite::SQLITE_TOOBIG
367        }
368    } else if let Some(value) = number.as_f64() {
369        unsafe { sqlite::sqlite3_bind_double(stmt, parameter, value) }
370    } else {
371        sqlite::SQLITE_MISMATCH
372    }
373}
374
375fn bind_text(stmt: *mut sqlite::sqlite3_stmt, parameter: c_int, value: &str) -> Result<c_int> {
376    let value = CString::new(value)
377        .map_err(|_| SyncularError::config("query string param contains interior NUL byte"))?;
378    Ok(unsafe {
379        sqlite::sqlite3_bind_text(
380            stmt,
381            parameter,
382            value.as_ptr(),
383            value.as_bytes().len() as c_int,
384            sqlite::SQLITE_TRANSIENT(),
385        )
386    })
387}
388
389fn read_rows(db: &SqliteDb, stmt: &SqliteStatement) -> Result<Vec<Value>> {
390    let column_count = unsafe { sqlite::sqlite3_column_count(stmt.raw) };
391    let mut rows = Vec::new();
392
393    loop {
394        match unsafe { sqlite::sqlite3_step(stmt.raw) } {
395            sqlite::SQLITE_ROW => rows.push(read_row(stmt.raw, column_count)?),
396            sqlite::SQLITE_DONE => break,
397            _ => return Err(sqlite_message(db.raw, "step read-only query")),
398        }
399    }
400
401    Ok(rows)
402}
403
404fn read_row(stmt: *mut sqlite::sqlite3_stmt, column_count: c_int) -> Result<Value> {
405    let mut row = Map::new();
406    for column in 0..column_count {
407        let name = column_name(stmt, column)?;
408        let value = column_value(stmt, column)?;
409        row.insert(name, value);
410    }
411    Ok(Value::Object(row))
412}
413
414fn column_name(stmt: *mut sqlite::sqlite3_stmt, column: c_int) -> Result<String> {
415    let raw = unsafe { sqlite::sqlite3_column_name(stmt, column) };
416    if raw.is_null() {
417        return Err(SyncularError::storage(anyhow::anyhow!(
418            "SQLite returned a null column name"
419        )));
420    }
421    Ok(unsafe { CStr::from_ptr(raw) }
422        .to_string_lossy()
423        .into_owned())
424}
425
426fn column_value(stmt: *mut sqlite::sqlite3_stmt, column: c_int) -> Result<Value> {
427    match unsafe { sqlite::sqlite3_column_type(stmt, column) } {
428        sqlite::SQLITE_NULL => Ok(Value::Null),
429        sqlite::SQLITE_INTEGER => Ok(Value::Number(Number::from(unsafe {
430            sqlite::sqlite3_column_int64(stmt, column)
431        }))),
432        sqlite::SQLITE_FLOAT => {
433            let value = unsafe { sqlite::sqlite3_column_double(stmt, column) };
434            Ok(Number::from_f64(value).map_or(Value::Null, Value::Number))
435        }
436        sqlite::SQLITE_TEXT => {
437            let raw = unsafe { sqlite::sqlite3_column_text(stmt, column) };
438            let len = unsafe { sqlite::sqlite3_column_bytes(stmt, column) };
439            if raw.is_null() {
440                return Ok(Value::Null);
441            }
442            let bytes = unsafe { slice::from_raw_parts(raw, len as usize) };
443            let value = std::str::from_utf8(bytes)
444                .map_err(|error| SyncularError::storage(anyhow::anyhow!(error)))?;
445            Ok(Value::String(value.to_string()))
446        }
447        sqlite::SQLITE_BLOB => {
448            let raw = unsafe { sqlite::sqlite3_column_blob(stmt, column) };
449            let len = unsafe { sqlite::sqlite3_column_bytes(stmt, column) };
450            if raw.is_null() {
451                return Ok(Value::Null);
452            }
453            let bytes = unsafe { slice::from_raw_parts(raw.cast::<u8>(), len as usize) };
454            Ok(json_blob(bytes))
455        }
456        _ => Err(SyncularError::storage(anyhow::anyhow!(
457            "unknown SQLite column type"
458        ))),
459    }
460}
461
462fn json_blob(bytes: &[u8]) -> Value {
463    let mut value = Map::new();
464    value.insert(
465        "$syncularType".to_string(),
466        Value::String("blob".to_string()),
467    );
468    value.insert("hex".to_string(), Value::String(hex::encode(bytes)));
469    Value::Object(value)
470}
471
472fn sqlite_message(db: *mut sqlite::sqlite3, action: &str) -> SyncularError {
473    if db.is_null() {
474        return SyncularError::storage(anyhow::anyhow!("{action} failed"));
475    }
476    let message = unsafe { CStr::from_ptr(sqlite::sqlite3_errmsg(db)) }
477        .to_string_lossy()
478        .into_owned();
479    SyncularError::storage(anyhow::anyhow!("{action} failed: {message}"))
480}
481
482unsafe extern "C" fn readonly_authorizer(
483    user_data: *mut c_void,
484    action: c_int,
485    arg1: *const c_char,
486    _arg2: *const c_char,
487    _database: *const c_char,
488    _trigger: *const c_char,
489) -> c_int {
490    let context = &mut *(user_data as *mut AuthorizerContext);
491    match action {
492        sqlite::SQLITE_SELECT | sqlite::SQLITE_FUNCTION => sqlite::SQLITE_OK,
493        sqlite::SQLITE_READ => {
494            let table = cstr_arg(arg1);
495            if table
496                .as_deref()
497                .is_some_and(|table| context.allowed_tables.contains(table))
498            {
499                sqlite::SQLITE_OK
500            } else {
501                context.denied = Some(match table {
502                    Some(table) => format!(
503                        "queryJson can only read declared generated app tables; denied table: {table}"
504                    ),
505                    None => "queryJson denied an unreadable table reference".to_string(),
506                });
507                sqlite::SQLITE_DENY
508            }
509        }
510        _ => {
511            context.denied = Some(format!(
512                "queryJson only allows read-only SELECT statements; denied SQLite action {action}"
513            ));
514            sqlite::SQLITE_DENY
515        }
516    }
517}
518
519fn cstr_arg(raw: *const c_char) -> Option<String> {
520    if raw.is_null() {
521        return None;
522    }
523    Some(
524        unsafe { CStr::from_ptr(raw) }
525            .to_string_lossy()
526            .into_owned(),
527    )
528}
syncular_runtime/storage/sqlite_query.rs

syncular_runtime/storage/
sqlite_query.rs
