dactyl_db/
lib.rs

1//! Dactyl — the governed datastore boundary for Decapod.
2//!
3//! Interchangeably read and write to local SQLite or cloud-hosted Vercel Neon
4//! instances behind a single unified facade.
5
6pub mod adapter;
7pub mod error;
8pub mod query;
9mod rows;
10
11#[doc(hidden)]
12pub mod __private;
13
14pub use dactyl_db_macros::query;
15
16pub use crate::error::DactylError;
17pub use crate::rows::{Parameter, Row, Rows};
18
19use std::sync::{Arc, Mutex, OnceLock};
20
21use crate::adapter::Adapter;
22
23/// A parameterized SQL statement for batch execution.
24#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq)]
25pub struct Statement {
26    pub sql: String,
27    pub params: Vec<Parameter>,
28}
29
30impl Statement {
31    /// Construct a new parameter-bound statement.
32    pub fn new(sql: &str, params: Vec<Parameter>) -> Self {
33        Self {
34            sql: sql.to_string(),
35            params,
36        }
37    }
38}
39
40/// Lazy connections, keyed by the connection string. Populated by the first
41/// datastore call for a given key.
42static CONNECTIONS: OnceLock<Mutex<std::collections::HashMap<String, Arc<dyn Adapter>>>> =
43    OnceLock::new();
44
45fn connections() -> &'static Mutex<std::collections::HashMap<String, Arc<dyn Adapter>>> {
46    CONNECTIONS.get_or_init(|| Mutex::new(std::collections::HashMap::new()))
47}
48
49/// Reset the cached connections. Test-only helper exposed to integration
50/// tests; production code should never call this.
51#[doc(hidden)]
52pub fn __reset_for_tests() {
53    let mut guard = connections()
54        .lock()
55        .expect("dactyl connection lock poisoned");
56    guard.clear();
57}
58
59/// Explicitly configure the active datastore and route.
60///
61/// Sets the `DATASTORE` and `DATASTORE_ROUTE` environment variables,
62/// and optionally `DATASTORE_TOKEN`.
63pub fn init(datastore: &str, route: &str, token: Option<&str>) {
64    std::env::set_var("DATASTORE", datastore);
65    std::env::set_var("DATASTORE_ROUTE", route);
66    if let Some(t) = token {
67        std::env::set_var("DATASTORE_TOKEN", t);
68    } else {
69        std::env::remove_var("DATASTORE_TOKEN");
70    }
71}
72
73/// Reset/clear all cached connections.
74pub fn reset() {
75    __reset_for_tests();
76}
77
78/// Execute a parameterized read query.
79pub fn read(query: &str, params: &[Parameter], optimize: bool) -> Result<Rows, DactylError> {
80    dispatch(query, params, optimize, false)
81}
82
83/// Execute a parameterized write query.
84pub fn write(query: &str, params: &[Parameter], optimize: bool) -> Result<Rows, DactylError> {
85    dispatch(query, params, optimize, true)
86}
87
88/// Execute a raw schema/DDL/migration operation.
89pub fn execute(query: &str, params: &[Parameter]) -> Result<u64, DactylError> {
90    validate_env()?;
91    let analyzer = query::QueryAnalyzer::new();
92    let analyzed = analyzer.analyze(query);
93    let key = connection_key().ok_or_else(|| {
94        DactylError::Adapter("no adapter configured: set DATASTORE and DATASTORE_ROUTE".into())
95    })?;
96    let adapter = connection(&key, query, &analyzed)?;
97    let sql = analyzed.rewrite.apply(query);
98    adapter.execute_raw(&sql, params)
99}
100
101/// Execute an atomic batch of statements.
102pub fn transaction(statements: &[Statement]) -> Result<Vec<Rows>, DactylError> {
103    if statements.is_empty() {
104        return Ok(Vec::new());
105    }
106    validate_env()?;
107    let analyzer = query::QueryAnalyzer::new();
108    // Analyze first query to decide which connection/dialect to use
109    let analyzed = analyzer.analyze(&statements[0].sql);
110    let key = connection_key().ok_or_else(|| {
111        DactylError::Adapter("no adapter configured: set DATASTORE and DATASTORE_ROUTE".into())
112    })?;
113    let adapter = connection(&key, &statements[0].sql, &analyzed)?;
114
115    // We apply rewrites to all statements in the batch
116    let rewritten: Vec<Statement> = statements
117        .iter()
118        .map(|s| {
119            let a = analyzer.analyze(&s.sql);
120            Statement {
121                sql: a.rewrite.apply(&s.sql),
122                params: s.params.clone(),
123            }
124        })
125        .collect();
126
127    adapter.execute_batch(&rewritten)
128}
129
130fn validate_env() -> Result<(), DactylError> {
131    if let Ok(ds) = std::env::var("DATASTORE") {
132        if ds != "sqlite" && ds != "neon" {
133            return Err(DactylError::Adapter(
134                "invalid DATASTORE value: must be 'sqlite' or 'neon'".to_string(),
135            ));
136        }
137        if std::env::var("DATASTORE_ROUTE").is_err() {
138            return Err(DactylError::Adapter(
139                "DATASTORE_ROUTE is required when DATASTORE is set".into(),
140            ));
141        }
142    } else {
143        let has_legacy = std::env::var("DACTYL_NEON_ENDPOINT").is_ok()
144            || std::env::var("DACTYL_SQLITE_PATH").is_ok();
145        if !has_legacy {
146            return Err(DactylError::Adapter(
147                "no adapter configured: set DATASTORE and DATASTORE_ROUTE".into(),
148            ));
149        }
150    }
151    Ok(())
152}
153
154fn dispatch(
155    query: &str,
156    params: &[Parameter],
157    optimize: bool,
158    write: bool,
159) -> Result<Rows, DactylError> {
160    validate_env()?;
161
162    let analyzer = query::QueryAnalyzer::new();
163    let analyzed = analyzer.analyze(query);
164
165    let inferred_dialect = infer_dialect(&analyzed);
166
167    let key = connection_key().ok_or_else(|| {
168        DactylError::Adapter("no adapter configured: set DATASTORE and DATASTORE_ROUTE".into())
169    })?;
170    let adapter = connection(&key, query, &analyzed)?;
171    if !optimize {
172        if let Some(c) = query::first_unsupported(&analyzed.constructs, inferred_dialect) {
173            return Err(DactylError::Unsupported { construct: c });
174        }
175    }
176
177    let sql = analyzed.rewrite.apply(query);
178    adapter.execute(&sql, params, optimize, write)
179}
180
181fn infer_dialect(analyzed: &query::Analyzed) -> query::Dialect {
182    if let Some(override_ds) = analyzed.inline_override {
183        return match override_ds {
184            "neon" => query::Dialect::Postgres,
185            _ => query::Dialect::Sqlite,
186        };
187    }
188    if let Ok(ds) = std::env::var("DATASTORE") {
189        return match ds.as_str() {
190            "neon" => query::Dialect::Postgres,
191            _ => query::Dialect::Sqlite,
192        };
193    }
194    if std::env::var("DACTYL_NEON_ENDPOINT").is_ok() {
195        query::Dialect::Postgres
196    } else {
197        query::Dialect::Sqlite
198    }
199}
200
201fn connection(
202    key: &str,
203    query: &str,
204    analyzed: &query::Analyzed,
205) -> Result<Arc<dyn Adapter>, DactylError> {
206    {
207        let guard = connections()
208            .lock()
209            .expect("dactyl connection lock poisoned");
210        if let Some(existing) = guard.get(key) {
211            return Ok(existing.clone());
212        }
213    }
214    let adapter = build_adapter(query, analyzed)?;
215    let mut guard = connections()
216        .lock()
217        .expect("dactyl connection lock poisoned");
218    if let Some(existing) = guard.get(key) {
219        return Ok(existing.clone());
220    }
221    guard.insert(key.to_string(), adapter.clone());
222    Ok(adapter)
223}
224
225fn connection_key() -> Option<String> {
226    if let Ok(ds) = std::env::var("DATASTORE") {
227        if let Ok(route) = std::env::var("DATASTORE_ROUTE") {
228            return Some(format!("{ds}:{route}"));
229        }
230    }
231    if let Ok(endpoint) = std::env::var("DACTYL_NEON_ENDPOINT") {
232        return Some(format!("neon:{endpoint}"));
233    }
234    if let Ok(path) = std::env::var("DACTYL_SQLITE_PATH") {
235        return Some(format!("sqlite:{path}"));
236    }
237    None
238}
239
240#[cfg(feature = "sqlite")]
241fn sqlite_adapter(path: &str) -> Result<Arc<dyn Adapter>, DactylError> {
242    use crate::adapter::sqlite::SqliteAdapter;
243    let adapter =
244        SqliteAdapter::open(path).map_err(|e| DactylError::Adapter(format!("sqlite open: {e}")))?;
245    Ok(Arc::new(adapter))
246}
247
248#[cfg(feature = "neon")]
249fn neon_adapter() -> Result<Arc<dyn Adapter>, DactylError> {
250    use crate::adapter::neon::NeonAdapter;
251    let (endpoint, bearer) = resolve_neon_config()?;
252    let adapter = NeonAdapter::new(&endpoint, bearer, None);
253    Ok(Arc::new(adapter))
254}
255
256#[cfg(not(feature = "sqlite"))]
257fn sqlite_adapter(_path: &str) -> Result<Arc<dyn Adapter>, DactylError> {
258    Err(DactylError::Adapter(
259        "sqlite adapter requested but `sqlite` feature is disabled".into(),
260    ))
261}
262
263#[cfg(not(feature = "neon"))]
264fn neon_adapter() -> Result<Arc<dyn Adapter>, DactylError> {
265    Err(DactylError::Adapter(
266        "neon adapter requested but `neon` feature is disabled".into(),
267    ))
268}
269
270#[cfg(feature = "neon")]
271fn resolve_neon_config() -> Result<(String, Option<String>), DactylError> {
272    if let Ok(ds) = std::env::var("DATASTORE") {
273        if ds == "neon" {
274            let route = std::env::var("DATASTORE_ROUTE")
275                .map_err(|_| DactylError::Adapter("DATASTORE_ROUTE not set".into()))?;
276            let token = std::env::var("DATASTORE_TOKEN")
277                .ok()
278                .or_else(|| std::env::var("DACTYL_NEON_BEARER").ok());
279            return Ok((route, token));
280        }
281    }
282    let endpoint = std::env::var("DACTYL_NEON_ENDPOINT")
283        .map_err(|_| DactylError::Adapter("DACTYL_NEON_ENDPOINT not set".into()))?;
284    let bearer = std::env::var("DACTYL_NEON_BEARER").ok();
285    Ok((endpoint, bearer))
286}
287
288fn resolve_sqlite_path(query: &str) -> Result<String, DactylError> {
289    if let Ok(ds) = std::env::var("DATASTORE") {
290        if ds == "sqlite" {
291            let route = std::env::var("DATASTORE_ROUTE")
292                .map_err(|_| DactylError::Adapter("DATASTORE_ROUTE not set".into()))?;
293            return Ok(route);
294        }
295    }
296    if let Ok(p) = std::env::var("DACTYL_SQLITE_PATH") {
297        return Ok(p);
298    }
299    let default_root =
300        std::env::var("DACTYL_SQLITE_ROOT").unwrap_or_else(|_| ".decapod/data".to_string());
301    let store = infer_store(query).unwrap_or_else(|| "dactyl".to_string());
302    Ok(format!("{default_root}/{store}.db"))
303}
304
305fn build_adapter(query: &str, analyzed: &query::Analyzed) -> Result<Arc<dyn Adapter>, DactylError> {
306    let neon_env = if let Ok(ds) = std::env::var("DATASTORE") {
307        ds == "neon"
308    } else {
309        std::env::var("DACTYL_NEON_ENDPOINT").is_ok()
310    };
311    let sqlite_only = !analyzed.constructs.is_empty()
312        && analyzed
313            .constructs
314            .iter()
315            .all(|c| c.dialect() == query::Dialect::Sqlite);
316
317    if neon_env && !sqlite_only {
318        neon_adapter()
319    } else {
320        sqlite_adapter(&resolve_sqlite_path(query)?)
321    }
322}
323
324fn infer_store(query: &str) -> Option<String> {
325    let lower = query.to_ascii_lowercase();
326    let mut iter = lower.split_whitespace();
327    while let Some(tok) = iter.next() {
328        if tok == "from" || tok == "into" || tok == "update" || tok == "table" {
329            if let Some(name) = iter.next() {
330                return Some(name.trim_end_matches(';').to_string());
331            }
332        }
333    }
334    None
335}
dactyl_db/lib.rs

dactyl_db/
lib.rs
