1use std::sync::Arc;
4
5use super::{
6 backend::DatabaseBackend,
7 error::Result,
8 query_builder::{DeleteBuilder, InsertBuilder, SelectBuilder, UpdateBuilder},
9};
10
11#[cfg(feature = "postgres")]
12use super::dialect::PostgresBackend;
13
14#[cfg(feature = "postgres")]
16const SQLSTATE_INVALID_CATALOG_NAME: &str = "3D000";
17
18#[cfg(feature = "sqlite")]
19use super::dialect::SqliteBackend;
20
21#[cfg(feature = "mysql")]
22use super::dialect::MySqlBackend;
23
24#[derive(Clone)]
26pub struct DatabaseConnection {
27 backend: Arc<dyn DatabaseBackend>,
28 is_cockroachdb: bool,
37}
38
39#[cfg(feature = "di")]
68#[async_trait::async_trait]
69impl reinhardt_di::Injectable for DatabaseConnection {
70 async fn inject(ctx: &reinhardt_di::InjectionContext) -> reinhardt_di::DiResult<Self> {
71 if let Some(conn) = ctx.get_singleton::<Self>() {
73 return Ok(std::sync::Arc::try_unwrap(conn).unwrap_or_else(|arc| (*arc).clone()));
74 }
75
76 if let Some(conn) = ctx.get_request::<Self>() {
78 return Ok(std::sync::Arc::try_unwrap(conn).unwrap_or_else(|arc| (*arc).clone()));
79 }
80
81 Err(reinhardt_di::DiError::NotRegistered {
83 type_name: std::any::type_name::<Self>().to_string(),
84 hint: "Use InjectionContextBuilder::singleton(db_connection) to register a \
85 DatabaseConnection. Create it with DatabaseConnection::connect_postgres(), \
86 connect_sqlite(), or connect_mysql()."
87 .to_string(),
88 })
89 }
90
91 async fn inject_uncached(ctx: &reinhardt_di::InjectionContext) -> reinhardt_di::DiResult<Self> {
92 Self::inject(ctx).await
95 }
96}
97
98impl DatabaseConnection {
99 pub fn new(backend: Arc<dyn DatabaseBackend>) -> Self {
107 Self::new_with_flavor(backend, false)
108 }
109
110 pub fn new_with_flavor(backend: Arc<dyn DatabaseBackend>, is_cockroachdb: bool) -> Self {
116 Self {
117 backend,
118 is_cockroachdb,
119 }
120 }
121
122 #[cfg(feature = "postgres")]
123 pub async fn connect_postgres(url: &str) -> Result<Self> {
125 Self::connect_postgres_with_pool_size(url, None).await
126 }
127
128 #[cfg(feature = "postgres")]
129 pub async fn connect_postgres_with_pool_size(
131 url: &str,
132 pool_size: Option<u32>,
133 ) -> Result<Self> {
134 let pool = Self::build_postgres_pool(url, pool_size).await?;
135 let is_cockroachdb = Self::probe_cockroachdb(&pool).await;
136
137 Ok(Self {
138 backend: Arc::new(PostgresBackend::new(pool)),
139 is_cockroachdb,
140 })
141 }
142
143 #[cfg(feature = "postgres")]
158 async fn probe_cockroachdb(pool: &sqlx::PgPool) -> bool {
159 sqlx::query_scalar::<_, bool>("SELECT version() LIKE 'CockroachDB%'")
160 .fetch_one(pool)
161 .await
162 .unwrap_or(false)
163 }
164
165 #[cfg(feature = "postgres")]
191 pub async fn connect_postgres_or_create(url: &str) -> Result<Self> {
192 Self::connect_postgres_or_create_with_pool_size(url, None).await
193 }
194
195 #[cfg(feature = "postgres")]
200 async fn build_postgres_pool(
201 url: &str,
202 pool_size: Option<u32>,
203 ) -> std::result::Result<sqlx::PgPool, sqlx::Error> {
204 use sqlx::postgres::PgPoolOptions;
205 use std::time::Duration;
206
207 let max_connections = pool_size
209 .or_else(|| {
210 std::env::var("DATABASE_POOL_MAX_CONNECTIONS")
211 .ok()
212 .and_then(|v| v.parse::<u32>().ok())
213 })
214 .unwrap_or(20); PgPoolOptions::new()
217 .max_connections(max_connections)
218 .min_connections(1) .acquire_timeout(Duration::from_secs(10)) .idle_timeout(Some(Duration::from_secs(10))) .max_lifetime(Some(Duration::from_secs(30 * 60))) .connect(url)
223 .await
224 }
225
226 #[cfg(feature = "postgres")]
230 pub async fn connect_postgres_or_create_with_pool_size(
231 url: &str,
232 pool_size: Option<u32>,
233 ) -> Result<Self> {
234 match Self::build_postgres_pool(url, pool_size).await {
237 Ok(pool) => {
238 let is_cockroachdb = Self::probe_cockroachdb(&pool).await;
239 return Ok(Self {
240 backend: Arc::new(PostgresBackend::new(pool)),
241 is_cockroachdb,
242 });
243 }
244 Err(e) => {
245 let is_db_not_found = matches!(
248 &e,
249 sqlx::Error::Database(db_err) if db_err.code().as_deref() == Some(SQLSTATE_INVALID_CATALOG_NAME)
250 );
251 if !is_db_not_found {
252 return Err(e.into());
253 }
254 }
256 }
257
258 let (admin_url, db_name) = Self::parse_postgres_url_for_creation(url)?;
260
261 use sqlx::postgres::PgPoolOptions;
263 use std::time::Duration;
264
265 let admin_pool = PgPoolOptions::new()
266 .max_connections(1)
267 .acquire_timeout(Duration::from_secs(10))
268 .connect(&admin_url)
269 .await
270 .map_err(|e| {
271 super::error::DatabaseError::ConnectionError(format!(
272 "Failed to connect to postgres database for auto-creation: {}",
273 e
274 ))
275 })?;
276
277 let create_sql = format!("CREATE DATABASE \"{}\"", db_name.replace('"', "\"\""));
279 sqlx::query(&create_sql)
280 .execute(&admin_pool)
281 .await
282 .map_err(|e| {
283 super::error::DatabaseError::QueryError(format!(
284 "Failed to create database '{}': {}",
285 db_name, e
286 ))
287 })?;
288
289 admin_pool.close().await;
291
292 Self::connect_postgres_with_pool_size(url, pool_size).await
294 }
295
296 #[cfg(feature = "postgres")]
298 fn parse_postgres_url_for_creation(url: &str) -> Result<(String, String)> {
299 let url_without_scheme = url
304 .strip_prefix("postgres://")
305 .or_else(|| url.strip_prefix("postgresql://"))
306 .ok_or_else(|| {
307 super::error::DatabaseError::ConnectionError(
308 "Invalid PostgreSQL URL: must start with postgres:// or postgresql://"
309 .to_string(),
310 )
311 })?;
312
313 let (path_part, query_part) = match url_without_scheme.find('?') {
315 Some(pos) => (&url_without_scheme[..pos], Some(&url_without_scheme[pos..])),
316 None => (url_without_scheme, None),
317 };
318
319 let last_slash_pos = path_part.rfind('/').ok_or_else(|| {
321 super::error::DatabaseError::ConnectionError(
322 "Invalid PostgreSQL URL: no database name found".to_string(),
323 )
324 })?;
325
326 let host_part = &path_part[..last_slash_pos];
327 let db_name = &path_part[last_slash_pos + 1..];
328
329 if db_name.is_empty() {
330 return Err(super::error::DatabaseError::ConnectionError(
331 "Invalid PostgreSQL URL: database name is empty".to_string(),
332 ));
333 }
334
335 let admin_url = match query_part {
337 Some(params) => format!("postgres://{}/postgres{}", host_part, params),
338 None => format!("postgres://{}/postgres", host_part),
339 };
340
341 Ok((admin_url, db_name.to_string()))
342 }
343
344 #[cfg(feature = "sqlite")]
346 pub async fn connect_sqlite(url: &str) -> Result<Self> {
347 use sqlx::sqlite::{SqliteConnectOptions, SqlitePool};
348 use std::path::Path;
349 use std::str::FromStr;
350
351 if url == "sqlite::memory:" {
353 let pool = SqlitePool::connect(url).await?;
354 return Ok(Self {
355 backend: Arc::new(SqliteBackend::new(pool)),
356 is_cockroachdb: false,
357 });
358 }
359
360 let file_path = if url.starts_with("sqlite:///") {
362 url.trim_start_matches("sqlite:///").to_string()
364 } else if url.starts_with("sqlite://") {
365 let rel_path = url.trim_start_matches("sqlite://");
368 std::env::current_dir()
369 .map_err(|e| {
370 super::error::DatabaseError::ConnectionError(format!(
371 "Failed to get current directory: {}",
372 e
373 ))
374 })?
375 .join(rel_path)
376 .to_string_lossy()
377 .to_string()
378 } else if url.starts_with("sqlite:") {
379 let rel_path = url.trim_start_matches("sqlite:");
382 std::env::current_dir()
383 .map_err(|e| {
384 super::error::DatabaseError::ConnectionError(format!(
385 "Failed to get current directory: {}",
386 e
387 ))
388 })?
389 .join(rel_path)
390 .to_string_lossy()
391 .to_string()
392 } else {
393 url.to_string()
394 };
395
396 let db_path = Path::new(&file_path);
398 let normalized_path = if db_path.exists() {
399 db_path.canonicalize().map_err(|e| {
401 super::error::DatabaseError::ConnectionError(format!(
402 "Failed to canonicalize path {}: {}",
403 db_path.display(),
404 e
405 ))
406 })?
407 } else {
408 if db_path.is_absolute() {
410 db_path.to_path_buf()
411 } else {
412 std::env::current_dir()
414 .map_err(|e| {
415 super::error::DatabaseError::ConnectionError(format!(
416 "Failed to get current directory: {}",
417 e
418 ))
419 })?
420 .join(db_path)
421 }
422 };
423
424 if let Some(parent) = normalized_path.parent()
426 && !parent.as_os_str().is_empty()
427 && !parent.exists()
428 {
429 std::fs::create_dir_all(parent).map_err(|e| {
430 super::error::DatabaseError::ConnectionError(format!(
431 "Failed to create database directory {}: {}",
432 parent.display(),
433 e
434 ))
435 })?;
436 }
437
438 let path_str = normalized_path.to_string_lossy().replace('\\', "/");
441 let absolute_url = format!("sqlite:///{}", path_str);
442
443 let options = SqliteConnectOptions::from_str(&absolute_url)
445 .map_err(|e| {
446 super::error::DatabaseError::ConnectionError(format!(
447 "Invalid SQLite URL '{}': {}",
448 absolute_url, e
449 ))
450 })?
451 .create_if_missing(true);
452
453 let pool = SqlitePool::connect_with(options).await?;
454
455 Ok(Self {
456 backend: Arc::new(SqliteBackend::new(pool)),
457 is_cockroachdb: false,
458 })
459 }
460
461 #[cfg(feature = "sqlite")]
463 pub fn from_sqlite_pool(pool: sqlx::SqlitePool) -> Self {
464 Self {
465 backend: Arc::new(SqliteBackend::new(pool)),
466 is_cockroachdb: false,
467 }
468 }
469
470 #[cfg(feature = "mysql")]
472 pub async fn connect_mysql(url: &str) -> Result<Self> {
473 use sqlx::MySqlPool;
474 let pool = MySqlPool::connect(url).await?;
475 Ok(Self {
476 backend: Arc::new(MySqlBackend::new(pool)),
477 is_cockroachdb: false,
478 })
479 }
480
481 pub fn backend(&self) -> Arc<dyn DatabaseBackend> {
483 self.backend.clone()
484 }
485
486 pub fn database_type(&self) -> super::types::DatabaseType {
488 self.backend.database_type()
489 }
490
491 pub fn is_cockroachdb(&self) -> bool {
502 self.is_cockroachdb
503 }
504
505 pub fn insert(&self, table: impl Into<String>) -> InsertBuilder {
507 InsertBuilder::new(self.backend.clone(), table)
508 }
509
510 pub fn update(&self, table: impl Into<String>) -> UpdateBuilder {
512 UpdateBuilder::new(self.backend.clone(), table)
513 }
514
515 pub fn select(&self) -> SelectBuilder {
517 SelectBuilder::new(self.backend.clone())
518 }
519
520 pub fn delete(&self, table: impl Into<String>) -> DeleteBuilder {
522 DeleteBuilder::new(self.backend.clone(), table)
523 }
524
525 #[cfg(feature = "settings")]
565 pub fn database_url_from<S>(settings: &S, env_override: Option<&str>) -> Result<String>
566 where
567 S: reinhardt_conf::HasCoreSettings + ?Sized,
568 {
569 if let Some(url) = env_override {
570 return Ok(url.to_string());
571 }
572
573 let core = settings.core();
574 let db_config = core.databases.get("default").ok_or_else(|| {
575 super::error::DatabaseError::ConnectionError(
576 "Database configuration `core.databases.default` not found in settings."
577 .to_string(),
578 )
579 })?;
580
581 Ok(db_config.to_url())
582 }
583
584 pub async fn execute(
586 &self,
587 sql: &str,
588 params: Vec<super::types::QueryValue>,
589 ) -> Result<super::types::QueryResult> {
590 self.backend.execute(sql, params).await
591 }
592
593 pub async fn fetch_one(
595 &self,
596 sql: &str,
597 params: Vec<super::types::QueryValue>,
598 ) -> Result<super::types::Row> {
599 self.backend.fetch_one(sql, params).await
600 }
601
602 pub async fn fetch_all(
604 &self,
605 sql: &str,
606 params: Vec<super::types::QueryValue>,
607 ) -> Result<Vec<super::types::Row>> {
608 self.backend.fetch_all(sql, params).await
609 }
610
611 pub async fn fetch_optional(
613 &self,
614 sql: &str,
615 params: Vec<super::types::QueryValue>,
616 ) -> Result<Option<super::types::Row>> {
617 self.backend.fetch_optional(sql, params).await
618 }
619
620 pub async fn begin(&self) -> Result<Box<dyn super::types::TransactionExecutor>> {
647 self.backend.begin().await
648 }
649
650 pub async fn begin_with_isolation(
668 &self,
669 level: super::types::IsolationLevel,
670 ) -> Result<Box<dyn super::types::TransactionExecutor>> {
671 self.backend.begin_with_isolation(level).await
672 }
673
674 #[cfg(feature = "postgres")]
675 pub fn into_postgres(&self) -> Option<sqlx::PgPool> {
677 self.backend
678 .as_any()
679 .downcast_ref::<super::dialect::PostgresBackend>()
680 .map(|backend| backend.pool().clone())
681 }
682
683 #[cfg(feature = "sqlite")]
685 pub fn into_sqlite(&self) -> Option<sqlx::SqlitePool> {
686 self.backend
687 .as_any()
688 .downcast_ref::<super::dialect::SqliteBackend>()
689 .map(|backend| backend.pool().clone())
690 }
691
692 #[cfg(feature = "mysql")]
694 pub fn into_mysql(&self) -> Option<sqlx::MySqlPool> {
695 self.backend
696 .as_any()
697 .downcast_ref::<super::dialect::MySqlBackend>()
698 .map(|backend| backend.pool().clone())
699 }
700}
701
702#[cfg(test)]
703mod tests {
704 use rstest::rstest;
705
706 fn build_create_database_sql(db_name: &str) -> String {
709 format!("CREATE DATABASE \"{}\"", db_name.replace('"', "\"\""))
710 }
711
712 #[rstest]
713 fn test_create_database_sql_normal_name() {
714 let db_name = "my_database";
716
717 let sql = build_create_database_sql(db_name);
719
720 assert_eq!(sql, "CREATE DATABASE \"my_database\"");
722 }
723
724 #[rstest]
725 fn test_create_database_sql_injection_with_double_quotes() {
726 let db_name = "test\"; DROP TABLE users; --";
728
729 let sql = build_create_database_sql(db_name);
731
732 assert_eq!(sql, "CREATE DATABASE \"test\"\"; DROP TABLE users; --\"");
734 }
737
738 #[rstest]
739 fn test_create_database_sql_injection_with_multiple_quotes() {
740 let db_name = "db\"\"injection";
742
743 let sql = build_create_database_sql(db_name);
745
746 assert_eq!(sql, "CREATE DATABASE \"db\"\"\"\"injection\"");
748 }
749
750 #[cfg(feature = "postgres")]
751 #[rstest]
752 fn test_parse_postgres_url_extracts_db_name() {
753 let url = "postgres://user:pass@localhost:5432/testdb";
755
756 let (admin_url, db_name) =
758 super::DatabaseConnection::parse_postgres_url_for_creation(url).unwrap();
759
760 assert_eq!(db_name, "testdb");
762 assert_eq!(admin_url, "postgres://user:pass@localhost:5432/postgres");
763 }
764}