nidus_sqlx/
lib.rs

1#![deny(missing_docs)]
2
3//! Official SQLx adapter for Nidus applications.
4//!
5//! This crate is installed separately from the core `nidus` facade so SQLx
6//! dependencies are only compiled by applications that choose this adapter.
7
8use nidus_core::NidusError;
9use thiserror::Error;
10
11/// Result type used by SQLx adapter operations.
12pub type Result<T> = std::result::Result<T, SqlxError>;
13
14/// Error returned by SQLx adapter operations.
15#[derive(Debug, Error)]
16pub enum SqlxError {
17    /// SQLx returned an error while building or checking a pool.
18    #[error(transparent)]
19    Sqlx(#[from] sqlx::Error),
20
21    /// Nidus provider registration failed.
22    #[error(transparent)]
23    Nidus(#[from] NidusError),
24
25    /// Nidus config deserialization failed.
26    #[cfg(feature = "nidus-config")]
27    #[error(transparent)]
28    Config(#[from] nidus_config::ConfigError),
29}
30
31#[cfg(any(feature = "mysql", feature = "cockroach"))]
32fn configuration_error(message: &'static str) -> SqlxError {
33    sqlx::Error::InvalidArgument(format!("invalid SQLx adapter configuration: {message}")).into()
34}
35
36#[cfg(feature = "sqlite")]
37mod sqlite {
38    #[cfg(feature = "observability")]
39    use std::time::Instant;
40
41    use super::Result;
42    use nidus_core::Container;
43
44    /// Typed configuration for a SQLx SQLite pool.
45    #[derive(Clone, Eq, PartialEq)]
46    pub struct SqlitePoolConfig {
47        database_url: String,
48        max_connections: Option<u32>,
49    }
50
51    impl SqlitePoolConfig {
52        /// Creates SQLite pool config from an explicit database URL.
53        pub fn new(database_url: impl Into<String>) -> Self {
54            Self {
55                database_url: database_url.into(),
56                max_connections: None,
57            }
58        }
59
60        /// Sets the maximum number of pool connections.
61        pub fn with_max_connections(mut self, max_connections: u32) -> Self {
62            self.max_connections = Some(max_connections);
63            self
64        }
65
66        /// Returns the configured database URL.
67        pub fn database_url(&self) -> &str {
68            &self.database_url
69        }
70
71        /// Returns the configured maximum connection count.
72        pub fn max_connections(&self) -> Option<u32> {
73            self.max_connections
74        }
75
76        /// Loads SQLite pool config from a nested `nidus_config::Config` path.
77        #[cfg(feature = "nidus-config")]
78        pub fn from_config_path<I, S>(config: &nidus_config::Config, path: I) -> Result<Self>
79        where
80            I: IntoIterator<Item = S>,
81            S: AsRef<str>,
82        {
83            #[derive(serde::Deserialize)]
84            struct RawConfig {
85                url: String,
86                max_connections: Option<u32>,
87            }
88
89            let raw: RawConfig = config.get_required_path_typed(path)?;
90            let mut settings = Self::new(raw.url);
91            if let Some(max_connections) = raw.max_connections {
92                settings = settings.with_max_connections(max_connections);
93            }
94            Ok(settings)
95        }
96    }
97
98    impl std::fmt::Debug for SqlitePoolConfig {
99        fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
100            formatter
101                .debug_struct("SqlitePoolConfig")
102                .field("database_url", &"<redacted>")
103                .field("max_connections", &self.max_connections)
104                .finish()
105        }
106    }
107
108    /// Builder for a SQLx SQLite pool provider.
109    #[derive(Clone, Debug)]
110    pub struct SqlitePoolBuilder {
111        config: SqlitePoolConfig,
112        #[cfg(feature = "observability")]
113        observer: Option<nidus_observability::ObservabilityAdapterObserver>,
114    }
115
116    impl SqlitePoolBuilder {
117        /// Creates a builder using `sqlite::memory:`.
118        pub fn new() -> Self {
119            Self {
120                config: SqlitePoolConfig::new("sqlite::memory:"),
121                #[cfg(feature = "observability")]
122                observer: None,
123            }
124        }
125
126        /// Replaces the builder config.
127        pub fn config(mut self, config: SqlitePoolConfig) -> Self {
128            self.config = config;
129            self
130        }
131
132        /// Sets the database URL.
133        pub fn database_url(mut self, database_url: impl Into<String>) -> Self {
134            self.config.database_url = database_url.into();
135            self
136        }
137
138        /// Sets the maximum number of pool connections.
139        pub fn max_connections(mut self, max_connections: u32) -> Self {
140            self.config.max_connections = Some(max_connections);
141            self
142        }
143
144        /// Instruments adapter-owned SQLx pool operations with Nidus observability.
145        #[cfg(feature = "observability")]
146        pub fn observability(
147            mut self,
148            observer: nidus_observability::ObservabilityAdapterObserver,
149        ) -> Self {
150            self.observer = Some(observer);
151            self
152        }
153
154        /// Connects and returns a provider wrapping the real SQLx pool.
155        pub async fn connect(self) -> Result<SqlitePoolProvider> {
156            #[cfg(feature = "observability")]
157            let observer = self.observer;
158            let mut options = sqlx::sqlite::SqlitePoolOptions::new();
159            if let Some(max_connections) = self.config.max_connections {
160                options = options.max_connections(max_connections);
161            }
162            #[cfg(feature = "observability")]
163            let started_at = Instant::now();
164            let pool = options.connect(&self.config.database_url).await;
165            #[cfg(feature = "observability")]
166            record_adapter_operation(
167                &observer,
168                "connect",
169                nidus_observability::OperationStatus::from(pool.is_ok()),
170                started_at,
171            );
172            let pool = pool?;
173            Ok(SqlitePoolProvider {
174                pool,
175                #[cfg(feature = "observability")]
176                observer,
177            })
178        }
179
180        /// Connects a provider and registers it as a Nidus singleton.
181        pub async fn register(self, container: &mut Container) -> Result<()> {
182            let provider = self.connect().await?;
183            container.register_singleton(provider)?;
184            Ok(())
185        }
186    }
187
188    impl Default for SqlitePoolBuilder {
189        fn default() -> Self {
190            Self::new()
191        }
192    }
193
194    /// Nidus provider wrapping a real SQLx SQLite pool.
195    #[derive(Clone, Debug)]
196    pub struct SqlitePoolProvider {
197        pool: sqlx::SqlitePool,
198        #[cfg(feature = "observability")]
199        observer: Option<nidus_observability::ObservabilityAdapterObserver>,
200    }
201
202    impl SqlitePoolProvider {
203        /// Creates a SQLite provider builder.
204        pub fn builder() -> SqlitePoolBuilder {
205            SqlitePoolBuilder::new()
206        }
207
208        /// Creates a provider from an existing SQLx SQLite pool.
209        pub fn from_pool(pool: sqlx::SqlitePool) -> Self {
210            Self {
211                pool,
212                #[cfg(feature = "observability")]
213                observer: None,
214            }
215        }
216
217        /// Returns direct access to the underlying SQLx pool.
218        pub fn pool(&self) -> &sqlx::SqlitePool {
219            &self.pool
220        }
221
222        /// Consumes the provider and returns the underlying SQLx pool.
223        pub fn into_pool(self) -> sqlx::SqlitePool {
224            self.pool
225        }
226
227        /// Executes a lightweight readiness query.
228        #[cfg(feature = "health")]
229        pub async fn health_status(&self) -> nidus_http::health::HealthStatus {
230            #[cfg(feature = "observability")]
231            let started_at = Instant::now();
232            let result = sqlx::query("SELECT 1").execute(&self.pool).await;
233            #[cfg(feature = "observability")]
234            record_adapter_operation(
235                &self.observer,
236                "health",
237                nidus_observability::OperationStatus::from(result.is_ok()),
238                started_at,
239            );
240            match result {
241                Ok(_) => nidus_http::health::HealthStatus::up(),
242                Err(error) => nidus_http::health::HealthStatus::down(error.to_string()),
243            }
244        }
245
246        /// Adds this provider as a readiness check on a health registry.
247        ///
248        /// The provider is expected to be the shared instance resolved from the
249        /// Nidus container, so the method takes `Arc<Self>` and does not clone
250        /// the underlying SQLx pool directly.
251        #[cfg(feature = "health")]
252        pub fn register_ready_check(
253            self: std::sync::Arc<Self>,
254            registry: nidus_http::health::HealthRegistry,
255            name: impl Into<String>,
256        ) -> nidus_http::health::HealthRegistry {
257            registry.ready_check(name, move || {
258                let provider = std::sync::Arc::clone(&self);
259                async move { provider.health_status().await }
260            })
261        }
262    }
263
264    #[async_trait::async_trait]
265    impl nidus_core::LifecycleHook for SqlitePoolProvider {
266        async fn on_shutdown(&self) -> nidus_core::Result<()> {
267            self.pool.close().await;
268            Ok(())
269        }
270    }
271
272    #[cfg(feature = "observability")]
273    fn record_adapter_operation(
274        observer: &Option<nidus_observability::ObservabilityAdapterObserver>,
275        operation: &'static str,
276        status: nidus_observability::OperationStatus,
277        started_at: Instant,
278    ) {
279        if let Some(observer) = observer {
280            observer.record("nidus-sqlx", operation, status, started_at.elapsed());
281        }
282    }
283}
284
285#[cfg(feature = "sqlite")]
286pub use sqlite::{SqlitePoolBuilder, SqlitePoolConfig, SqlitePoolProvider};
287
288#[cfg(feature = "postgres")]
289mod postgres {
290    #[cfg(feature = "observability")]
291    use std::time::Instant;
292
293    use super::Result;
294    use nidus_core::Container;
295
296    /// Typed configuration for a SQLx Postgres pool.
297    #[derive(Clone, Eq, PartialEq)]
298    pub struct PostgresPoolConfig {
299        database_url: String,
300        max_connections: Option<u32>,
301        min_connections: Option<u32>,
302    }
303
304    impl PostgresPoolConfig {
305        /// Creates Postgres pool config from an explicit database URL.
306        pub fn new(database_url: impl Into<String>) -> Self {
307            Self {
308                database_url: database_url.into(),
309                max_connections: None,
310                min_connections: None,
311            }
312        }
313
314        /// Sets the maximum number of pool connections.
315        pub fn with_max_connections(mut self, max_connections: u32) -> Self {
316            self.max_connections = Some(max_connections);
317            self
318        }
319
320        /// Sets the minimum number of pool connections.
321        pub fn with_min_connections(mut self, min_connections: u32) -> Self {
322            self.min_connections = Some(min_connections);
323            self
324        }
325
326        /// Returns the configured database URL.
327        pub fn database_url(&self) -> &str {
328            &self.database_url
329        }
330
331        /// Returns the configured maximum connection count.
332        pub fn max_connections(&self) -> Option<u32> {
333            self.max_connections
334        }
335
336        /// Returns the configured minimum connection count.
337        pub fn min_connections(&self) -> Option<u32> {
338            self.min_connections
339        }
340
341        /// Loads Postgres pool config from a nested `nidus_config::Config` path.
342        #[cfg(feature = "nidus-config")]
343        pub fn from_config_path<I, S>(config: &nidus_config::Config, path: I) -> Result<Self>
344        where
345            I: IntoIterator<Item = S>,
346            S: AsRef<str>,
347        {
348            #[derive(serde::Deserialize)]
349            struct RawConfig {
350                url: String,
351                max_connections: Option<u32>,
352                min_connections: Option<u32>,
353            }
354
355            let raw: RawConfig = config.get_required_path_typed(path)?;
356            let mut settings = Self::new(raw.url);
357            if let Some(max_connections) = raw.max_connections {
358                settings = settings.with_max_connections(max_connections);
359            }
360            if let Some(min_connections) = raw.min_connections {
361                settings = settings.with_min_connections(min_connections);
362            }
363            Ok(settings)
364        }
365    }
366
367    impl std::fmt::Debug for PostgresPoolConfig {
368        fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
369            formatter
370                .debug_struct("PostgresPoolConfig")
371                .field("database_url", &"<redacted>")
372                .field("max_connections", &self.max_connections)
373                .field("min_connections", &self.min_connections)
374                .finish()
375        }
376    }
377
378    /// Builder for a SQLx Postgres pool provider.
379    #[derive(Clone, Debug)]
380    pub struct PostgresPoolBuilder {
381        config: PostgresPoolConfig,
382        #[cfg(feature = "observability")]
383        observer: Option<nidus_observability::ObservabilityAdapterObserver>,
384    }
385
386    impl PostgresPoolBuilder {
387        /// Creates a builder using an explicit database URL.
388        pub fn new(database_url: impl Into<String>) -> Self {
389            Self {
390                config: PostgresPoolConfig::new(database_url),
391                #[cfg(feature = "observability")]
392                observer: None,
393            }
394        }
395
396        /// Replaces the builder config.
397        pub fn config(mut self, config: PostgresPoolConfig) -> Self {
398            self.config = config;
399            self
400        }
401
402        /// Sets the database URL.
403        pub fn database_url(mut self, database_url: impl Into<String>) -> Self {
404            self.config.database_url = database_url.into();
405            self
406        }
407
408        /// Sets the maximum number of pool connections.
409        pub fn max_connections(mut self, max_connections: u32) -> Self {
410            self.config.max_connections = Some(max_connections);
411            self
412        }
413
414        /// Sets the minimum number of pool connections.
415        pub fn min_connections(mut self, min_connections: u32) -> Self {
416            self.config.min_connections = Some(min_connections);
417            self
418        }
419
420        /// Instruments adapter-owned SQLx pool operations with Nidus observability.
421        #[cfg(feature = "observability")]
422        pub fn observability(
423            mut self,
424            observer: nidus_observability::ObservabilityAdapterObserver,
425        ) -> Self {
426            self.observer = Some(observer);
427            self
428        }
429
430        /// Connects and returns a provider wrapping the real SQLx pool.
431        pub async fn connect(self) -> Result<PostgresPoolProvider> {
432            #[cfg(feature = "observability")]
433            let observer = self.observer;
434            let mut options = sqlx::postgres::PgPoolOptions::new();
435            if let Some(max_connections) = self.config.max_connections {
436                options = options.max_connections(max_connections);
437            }
438            if let Some(min_connections) = self.config.min_connections {
439                options = options.min_connections(min_connections);
440            }
441            #[cfg(feature = "observability")]
442            let started_at = Instant::now();
443            let pool = options.connect(&self.config.database_url).await;
444            #[cfg(feature = "observability")]
445            record_adapter_operation(
446                &observer,
447                "connect",
448                nidus_observability::OperationStatus::from(pool.is_ok()),
449                started_at,
450            );
451            let pool = pool?;
452            Ok(PostgresPoolProvider {
453                pool,
454                #[cfg(feature = "observability")]
455                observer,
456            })
457        }
458
459        /// Connects a provider and registers it as a Nidus singleton.
460        pub async fn register(self, container: &mut Container) -> Result<()> {
461            let provider = self.connect().await?;
462            container.register_singleton(provider)?;
463            Ok(())
464        }
465    }
466
467    /// Nidus provider wrapping a real SQLx Postgres pool.
468    #[derive(Clone, Debug)]
469    pub struct PostgresPoolProvider {
470        pool: sqlx::PgPool,
471        #[cfg(feature = "observability")]
472        observer: Option<nidus_observability::ObservabilityAdapterObserver>,
473    }
474
475    impl PostgresPoolProvider {
476        /// Creates a Postgres provider builder.
477        pub fn builder(database_url: impl Into<String>) -> PostgresPoolBuilder {
478            PostgresPoolBuilder::new(database_url)
479        }
480
481        /// Creates a provider from an existing SQLx Postgres pool.
482        pub fn from_pool(pool: sqlx::PgPool) -> Self {
483            Self {
484                pool,
485                #[cfg(feature = "observability")]
486                observer: None,
487            }
488        }
489
490        /// Returns direct access to the underlying SQLx pool.
491        pub fn pool(&self) -> &sqlx::PgPool {
492            &self.pool
493        }
494
495        /// Consumes the provider and returns the underlying SQLx pool.
496        pub fn into_pool(self) -> sqlx::PgPool {
497            self.pool
498        }
499
500        /// Executes a lightweight readiness query.
501        #[cfg(feature = "health")]
502        pub async fn health_status(&self) -> nidus_http::health::HealthStatus {
503            #[cfg(feature = "observability")]
504            let started_at = Instant::now();
505            let result = sqlx::query("SELECT 1").execute(&self.pool).await;
506            #[cfg(feature = "observability")]
507            record_adapter_operation(
508                &self.observer,
509                "health",
510                nidus_observability::OperationStatus::from(result.is_ok()),
511                started_at,
512            );
513            match result {
514                Ok(_) => nidus_http::health::HealthStatus::up(),
515                Err(error) => nidus_http::health::HealthStatus::down(error.to_string()),
516            }
517        }
518
519        /// Adds this provider as a readiness check on a health registry.
520        ///
521        /// The provider is expected to be the shared instance resolved from the
522        /// Nidus container, so the method takes `Arc<Self>` and does not clone
523        /// the underlying SQLx pool directly.
524        #[cfg(feature = "health")]
525        pub fn register_ready_check(
526            self: std::sync::Arc<Self>,
527            registry: nidus_http::health::HealthRegistry,
528            name: impl Into<String>,
529        ) -> nidus_http::health::HealthRegistry {
530            registry.ready_check(name, move || {
531                let provider = std::sync::Arc::clone(&self);
532                async move { provider.health_status().await }
533            })
534        }
535    }
536
537    #[async_trait::async_trait]
538    impl nidus_core::LifecycleHook for PostgresPoolProvider {
539        async fn on_shutdown(&self) -> nidus_core::Result<()> {
540            self.pool.close().await;
541            Ok(())
542        }
543    }
544
545    #[cfg(feature = "observability")]
546    fn record_adapter_operation(
547        observer: &Option<nidus_observability::ObservabilityAdapterObserver>,
548        operation: &'static str,
549        status: nidus_observability::OperationStatus,
550        started_at: Instant,
551    ) {
552        if let Some(observer) = observer {
553            observer.record("nidus-sqlx", operation, status, started_at.elapsed());
554        }
555    }
556}
557
558#[cfg(feature = "postgres")]
559pub use postgres::{PostgresPoolBuilder, PostgresPoolConfig, PostgresPoolProvider};
560
561#[cfg(feature = "mysql")]
562mod mysql {
563    use std::{str::FromStr, time::Instant};
564
565    use nidus_core::Container;
566    use nidus_integrations::{IntegrationEvent, IntegrationStatus, IntegrationTelemetry};
567
568    use super::Result;
569
570    /// Typed configuration for a SQLx MySQL pool.
571    #[derive(Clone, Eq, PartialEq)]
572    pub struct MySqlPoolConfig {
573        database_url: String,
574        max_connections: Option<u32>,
575        min_connections: Option<u32>,
576        allow_insecure_local: bool,
577    }
578
579    impl MySqlPoolConfig {
580        /// Creates MySQL pool config from an explicit database URL.
581        pub fn new(database_url: impl Into<String>) -> Self {
582            Self {
583                database_url: database_url.into(),
584                max_connections: None,
585                min_connections: None,
586                allow_insecure_local: false,
587            }
588        }
589
590        /// Sets the maximum number of pool connections.
591        pub fn with_max_connections(mut self, max_connections: u32) -> Self {
592            self.max_connections = Some(max_connections);
593            self
594        }
595
596        /// Sets the minimum number of pool connections.
597        pub fn with_min_connections(mut self, min_connections: u32) -> Self {
598            self.min_connections = Some(min_connections);
599            self
600        }
601
602        /// Explicitly permits non-verifying TLS only for loopback development.
603        pub fn allow_insecure_for_local_development(mut self) -> Self {
604            self.allow_insecure_local = true;
605            self
606        }
607
608        /// Returns the configured database URL.
609        pub fn database_url(&self) -> &str {
610            &self.database_url
611        }
612
613        /// Returns the configured maximum connection count.
614        pub fn max_connections(&self) -> Option<u32> {
615            self.max_connections
616        }
617
618        /// Returns the configured minimum connection count.
619        pub fn min_connections(&self) -> Option<u32> {
620            self.min_connections
621        }
622
623        /// Validates pool bounds and TLS hostname verification.
624        pub fn validate(&self) -> Result<()> {
625            if self.max_connections == Some(0) {
626                return Err(super::configuration_error(
627                    "MySQL max_connections must be greater than zero",
628                ));
629            }
630            if let (Some(minimum), Some(maximum)) = (self.min_connections, self.max_connections)
631                && minimum > maximum
632            {
633                return Err(super::configuration_error(
634                    "MySQL min_connections cannot exceed max_connections",
635                ));
636            }
637            let options = sqlx::mysql::MySqlConnectOptions::from_str(&self.database_url)?;
638            let verifies_identity = matches!(
639                options.get_ssl_mode(),
640                sqlx::mysql::MySqlSslMode::VerifyIdentity
641            );
642            if !verifies_identity
643                && (!self.allow_insecure_local
644                    || !matches!(options.get_host(), "localhost" | "127.0.0.1" | "::1"))
645            {
646                return Err(super::configuration_error(
647                    "MySQL requires ssl-mode=VERIFY_IDENTITY except for explicit loopback development",
648                ));
649            }
650            Ok(())
651        }
652
653        /// Loads MySQL pool config from a nested `nidus_config::Config` path.
654        #[cfg(feature = "nidus-config")]
655        pub fn from_config_path<I, S>(config: &nidus_config::Config, path: I) -> Result<Self>
656        where
657            I: IntoIterator<Item = S>,
658            S: AsRef<str>,
659        {
660            #[derive(serde::Deserialize)]
661            struct RawConfig {
662                url: String,
663                max_connections: Option<u32>,
664                min_connections: Option<u32>,
665                allow_insecure_local: Option<bool>,
666            }
667
668            let raw: RawConfig = config.get_required_path_typed(path)?;
669            let mut settings = Self::new(raw.url);
670            if let Some(value) = raw.max_connections {
671                settings = settings.with_max_connections(value);
672            }
673            if let Some(value) = raw.min_connections {
674                settings = settings.with_min_connections(value);
675            }
676            if raw.allow_insecure_local == Some(true) {
677                settings = settings.allow_insecure_for_local_development();
678            }
679            settings.validate()?;
680            Ok(settings)
681        }
682    }
683
684    impl std::fmt::Debug for MySqlPoolConfig {
685        fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
686            formatter
687                .debug_struct("MySqlPoolConfig")
688                .field("database_url", &"<redacted>")
689                .field("max_connections", &self.max_connections)
690                .field("min_connections", &self.min_connections)
691                .field("allow_insecure_local", &self.allow_insecure_local)
692                .finish()
693        }
694    }
695
696    /// Builder for a SQLx MySQL pool provider.
697    #[derive(Clone, Debug)]
698    pub struct MySqlPoolBuilder {
699        config: MySqlPoolConfig,
700        telemetry: IntegrationTelemetry,
701    }
702
703    impl MySqlPoolBuilder {
704        /// Creates a builder using an explicit database URL.
705        pub fn new(database_url: impl Into<String>) -> Self {
706            Self {
707                config: MySqlPoolConfig::new(database_url),
708                telemetry: IntegrationTelemetry::new(),
709            }
710        }
711
712        /// Replaces the typed pool config.
713        pub fn config(mut self, config: MySqlPoolConfig) -> Self {
714            self.config = config;
715            self
716        }
717
718        /// Sets the database URL.
719        pub fn database_url(mut self, database_url: impl Into<String>) -> Self {
720            self.config.database_url = database_url.into();
721            self
722        }
723
724        /// Sets the maximum number of pool connections.
725        pub fn max_connections(mut self, max_connections: u32) -> Self {
726            self.config.max_connections = Some(max_connections);
727            self
728        }
729
730        /// Sets the minimum number of pool connections.
731        pub fn min_connections(mut self, min_connections: u32) -> Self {
732            self.config.min_connections = Some(min_connections);
733            self
734        }
735
736        /// Adds shared tracing, metrics, dashboard, or custom telemetry.
737        pub fn telemetry(mut self, telemetry: IntegrationTelemetry) -> Self {
738            self.telemetry = telemetry;
739            self
740        }
741
742        /// Instruments operations with an existing Nidus observability observer.
743        #[cfg(feature = "observability")]
744        pub fn observability(
745            mut self,
746            observer: nidus_observability::ObservabilityAdapterObserver,
747        ) -> Self {
748            self.telemetry = self.telemetry.observability(observer);
749            self
750        }
751
752        /// Connects and returns a provider wrapping the native SQLx pool.
753        pub async fn connect(self) -> Result<MySqlPoolProvider> {
754            self.config.validate()?;
755            let started_at = Instant::now();
756            let mut options = sqlx::mysql::MySqlPoolOptions::new();
757            if let Some(value) = self.config.max_connections {
758                options = options.max_connections(value);
759            }
760            if let Some(value) = self.config.min_connections {
761                options = options.min_connections(value);
762            }
763            let pool = options.connect(&self.config.database_url).await;
764            self.telemetry
765                .record(&IntegrationEvent::new(
766                    "nidus-sqlx-mysql",
767                    "connect",
768                    if pool.is_ok() {
769                        IntegrationStatus::Success
770                    } else {
771                        IntegrationStatus::Failure
772                    },
773                    started_at.elapsed(),
774                ))
775                .await;
776            Ok(MySqlPoolProvider {
777                pool: pool?,
778                telemetry: self.telemetry,
779            })
780        }
781
782        /// Connects and registers the provider as a Nidus singleton.
783        pub async fn register(self, container: &mut Container) -> Result<()> {
784            container.register_singleton(self.connect().await?)?;
785            Ok(())
786        }
787    }
788
789    /// Nidus provider wrapping a native SQLx MySQL pool.
790    #[derive(Clone, Debug)]
791    pub struct MySqlPoolProvider {
792        pool: sqlx::MySqlPool,
793        telemetry: IntegrationTelemetry,
794    }
795
796    impl MySqlPoolProvider {
797        /// Creates a MySQL provider builder.
798        pub fn builder(database_url: impl Into<String>) -> MySqlPoolBuilder {
799            MySqlPoolBuilder::new(database_url)
800        }
801
802        /// Creates a provider from an existing SQLx MySQL pool.
803        pub fn from_pool(pool: sqlx::MySqlPool) -> Self {
804            Self {
805                pool,
806                telemetry: IntegrationTelemetry::new(),
807            }
808        }
809
810        /// Returns direct access to the native SQLx pool.
811        pub fn pool(&self) -> &sqlx::MySqlPool {
812            &self.pool
813        }
814
815        /// Returns the shared adapter telemetry configuration.
816        pub const fn telemetry(&self) -> &IntegrationTelemetry {
817            &self.telemetry
818        }
819
820        /// Consumes the provider and returns the native SQLx pool.
821        pub fn into_pool(self) -> sqlx::MySqlPool {
822            self.pool
823        }
824
825        /// Executes a lightweight readiness query with a safe error response.
826        #[cfg(feature = "health")]
827        pub async fn health_status(&self) -> nidus_http::health::HealthStatus {
828            let started_at = Instant::now();
829            let result = sqlx::query("SELECT 1").execute(&self.pool).await;
830            self.telemetry
831                .record(&IntegrationEvent::new(
832                    "nidus-sqlx-mysql",
833                    "health",
834                    if result.is_ok() {
835                        IntegrationStatus::Success
836                    } else {
837                        IntegrationStatus::Failure
838                    },
839                    started_at.elapsed(),
840                ))
841                .await;
842            if result.is_ok() {
843                nidus_http::health::HealthStatus::up()
844            } else {
845                nidus_http::health::HealthStatus::down("mysql readiness check failed")
846            }
847        }
848
849        /// Adds this provider as a readiness check on a health registry.
850        #[cfg(feature = "health")]
851        pub fn register_ready_check(
852            self: std::sync::Arc<Self>,
853            registry: nidus_http::health::HealthRegistry,
854            name: impl Into<String>,
855        ) -> nidus_http::health::HealthRegistry {
856            registry.ready_check(name, move || {
857                let provider = std::sync::Arc::clone(&self);
858                async move { provider.health_status().await }
859            })
860        }
861    }
862
863    #[async_trait::async_trait]
864    impl nidus_core::LifecycleHook for MySqlPoolProvider {
865        async fn on_shutdown(&self) -> nidus_core::Result<()> {
866            self.pool.close().await;
867            Ok(())
868        }
869    }
870}
871
872#[cfg(feature = "mysql")]
873pub use mysql::{MySqlPoolBuilder, MySqlPoolConfig, MySqlPoolProvider};
874
875#[cfg(feature = "cockroach")]
876mod cockroach {
877    use std::{
878        fmt,
879        str::FromStr,
880        time::{Duration, Instant},
881    };
882
883    use futures_util::future::BoxFuture;
884    use nidus_core::Container;
885    use nidus_integrations::{IntegrationEvent, IntegrationStatus, IntegrationTelemetry};
886    use sqlx::Acquire;
887
888    use super::Result;
889
890    /// Result returned by a retryable CockroachDB transaction.
891    pub type CockroachTransactionResult<T> = std::result::Result<T, CockroachTransactionError>;
892
893    /// Error returned by a retryable CockroachDB transaction.
894    #[derive(Debug, thiserror::Error)]
895    pub enum CockroachTransactionError {
896        /// SQLx returned a non-retryable error.
897        #[error(transparent)]
898        Sqlx(#[from] sqlx::Error),
899        /// CockroachDB kept returning SQLSTATE `40001` through the attempt bound.
900        #[error("CockroachDB transaction exhausted {attempts} attempts")]
901        RetryExhausted {
902            /// Total transaction attempts.
903            attempts: usize,
904            /// Last retryable SQLx error.
905            #[source]
906            source: sqlx::Error,
907        },
908    }
909
910    /// Bounded application-level retry policy for CockroachDB serialization failures.
911    #[derive(Clone, Debug, Eq, PartialEq)]
912    pub struct CockroachRetryPolicy {
913        max_attempts: usize,
914        initial_backoff: Duration,
915        max_backoff: Duration,
916        jitter: bool,
917    }
918
919    impl CockroachRetryPolicy {
920        /// Creates the default bounded policy: five attempts, 25ms to 2s backoff, full jitter.
921        pub fn new() -> Self {
922            Self {
923                max_attempts: 5,
924                initial_backoff: Duration::from_millis(25),
925                max_backoff: Duration::from_secs(2),
926                jitter: true,
927            }
928        }
929
930        /// Sets the total number of attempts, including the first execution.
931        pub fn with_max_attempts(mut self, max_attempts: usize) -> Self {
932            self.max_attempts = max_attempts;
933            self
934        }
935
936        /// Sets the initial and maximum exponential backoff.
937        pub fn with_backoff(mut self, initial: Duration, maximum: Duration) -> Self {
938            self.initial_backoff = initial;
939            self.max_backoff = maximum;
940            self
941        }
942
943        /// Disables jitter for deterministic tests.
944        pub fn without_jitter(mut self) -> Self {
945            self.jitter = false;
946            self
947        }
948
949        /// Returns the total allowed transaction attempts.
950        pub const fn max_attempts(&self) -> usize {
951            self.max_attempts
952        }
953
954        /// Returns the maximum delay before a one-based retry attempt.
955        pub fn maximum_delay_for_retry(&self, retry: usize) -> Duration {
956            let shift = retry.saturating_sub(1).min(31) as u32;
957            self.initial_backoff
958                .saturating_mul(1_u32 << shift)
959                .min(self.max_backoff)
960        }
961
962        fn validate(&self) -> Result<()> {
963            if self.max_attempts == 0 {
964                return Err(super::configuration_error(
965                    "CockroachDB max_attempts must be greater than zero",
966                ));
967            }
968            if self.initial_backoff.is_zero() || self.max_backoff < self.initial_backoff {
969                return Err(super::configuration_error(
970                    "CockroachDB retry backoff must be positive and ordered",
971                ));
972            }
973            Ok(())
974        }
975
976        fn delay_for_retry(&self, retry: usize) -> Duration {
977            let maximum = self.maximum_delay_for_retry(retry);
978            if !self.jitter {
979                return maximum;
980            }
981            let upper = maximum.as_nanos().min(u128::from(u64::MAX)) as u64;
982            Duration::from_nanos(fastrand::u64(0..=upper))
983        }
984    }
985
986    impl Default for CockroachRetryPolicy {
987        fn default() -> Self {
988            Self::new()
989        }
990    }
991
992    /// Typed CockroachDB pool and retry configuration.
993    #[derive(Clone, Eq, PartialEq)]
994    pub struct CockroachPoolConfig {
995        database_url: String,
996        max_connections: Option<u32>,
997        min_connections: Option<u32>,
998        acquire_timeout: Duration,
999        allow_insecure_local: bool,
1000        retry_policy: CockroachRetryPolicy,
1001    }
1002
1003    impl CockroachPoolConfig {
1004        /// Creates CockroachDB config that requires `sslmode=verify-full`.
1005        pub fn new(database_url: impl Into<String>) -> Self {
1006            Self {
1007                database_url: database_url.into(),
1008                max_connections: None,
1009                min_connections: None,
1010                acquire_timeout: Duration::from_secs(5),
1011                allow_insecure_local: false,
1012                retry_policy: CockroachRetryPolicy::new(),
1013            }
1014        }
1015
1016        /// Sets the maximum number of pool connections.
1017        pub fn with_max_connections(mut self, value: u32) -> Self {
1018            self.max_connections = Some(value);
1019            self
1020        }
1021
1022        /// Sets the minimum number of pool connections.
1023        pub fn with_min_connections(mut self, value: u32) -> Self {
1024            self.min_connections = Some(value);
1025            self
1026        }
1027
1028        /// Sets the bounded pool acquisition timeout.
1029        pub fn with_acquire_timeout(mut self, value: Duration) -> Self {
1030            self.acquire_timeout = value;
1031            self
1032        }
1033
1034        /// Replaces the transaction retry policy.
1035        pub fn with_retry_policy(mut self, value: CockroachRetryPolicy) -> Self {
1036            self.retry_policy = value;
1037            self
1038        }
1039
1040        /// Explicitly permits non-TLS local development connections.
1041        ///
1042        /// Never enable this for deployed applications.
1043        pub fn allow_insecure_for_local_development(mut self) -> Self {
1044            self.allow_insecure_local = true;
1045            self
1046        }
1047
1048        /// Returns the configured database URL.
1049        pub fn database_url(&self) -> &str {
1050            &self.database_url
1051        }
1052
1053        /// Returns the transaction retry policy.
1054        pub fn retry_policy(&self) -> &CockroachRetryPolicy {
1055            &self.retry_policy
1056        }
1057
1058        /// Validates TLS and retry safety before any network I/O.
1059        pub fn validate(&self) -> Result<()> {
1060            self.retry_policy.validate()?;
1061            if self.max_connections == Some(0) {
1062                return Err(super::configuration_error(
1063                    "CockroachDB max_connections must be greater than zero",
1064                ));
1065            }
1066            if self.acquire_timeout.is_zero() {
1067                return Err(super::configuration_error(
1068                    "CockroachDB acquire timeout must be greater than zero",
1069                ));
1070            }
1071            if let (Some(minimum), Some(maximum)) = (self.min_connections, self.max_connections)
1072                && minimum > maximum
1073            {
1074                return Err(super::configuration_error(
1075                    "CockroachDB min_connections cannot exceed max_connections",
1076                ));
1077            }
1078            let options = sqlx::postgres::PgConnectOptions::from_str(&self.database_url)?;
1079            let verify_full = matches!(
1080                options.get_ssl_mode(),
1081                sqlx::postgres::PgSslMode::VerifyFull
1082            );
1083            if !verify_full
1084                && (!self.allow_insecure_local
1085                    || !matches!(options.get_host(), "localhost" | "127.0.0.1" | "::1"))
1086            {
1087                return Err(super::configuration_error(
1088                    "CockroachDB requires sslmode=verify-full except for explicit loopback development",
1089                ));
1090            }
1091            Ok(())
1092        }
1093
1094        /// Loads CockroachDB config from a nested `nidus_config::Config` path.
1095        #[cfg(feature = "nidus-config")]
1096        pub fn from_config_path<I, S>(config: &nidus_config::Config, path: I) -> Result<Self>
1097        where
1098            I: IntoIterator<Item = S>,
1099            S: AsRef<str>,
1100        {
1101            #[derive(serde::Deserialize)]
1102            struct RawConfig {
1103                url: String,
1104                max_connections: Option<u32>,
1105                min_connections: Option<u32>,
1106                acquire_timeout_ms: Option<u64>,
1107                max_attempts: Option<usize>,
1108                allow_insecure_local: Option<bool>,
1109            }
1110
1111            let raw: RawConfig = config.get_required_path_typed(path)?;
1112            let mut settings = Self::new(raw.url);
1113            if let Some(value) = raw.max_connections {
1114                settings = settings.with_max_connections(value);
1115            }
1116            if let Some(value) = raw.min_connections {
1117                settings = settings.with_min_connections(value);
1118            }
1119            if let Some(value) = raw.acquire_timeout_ms {
1120                settings = settings.with_acquire_timeout(Duration::from_millis(value));
1121            }
1122            if let Some(value) = raw.max_attempts {
1123                settings.retry_policy = settings.retry_policy.with_max_attempts(value);
1124            }
1125            if raw.allow_insecure_local == Some(true) {
1126                settings = settings.allow_insecure_for_local_development();
1127            }
1128            settings.validate()?;
1129            Ok(settings)
1130        }
1131    }
1132
1133    impl fmt::Debug for CockroachPoolConfig {
1134        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1135            formatter
1136                .debug_struct("CockroachPoolConfig")
1137                .field("database_url", &"<redacted>")
1138                .field("max_connections", &self.max_connections)
1139                .field("min_connections", &self.min_connections)
1140                .field("acquire_timeout", &self.acquire_timeout)
1141                .field("allow_insecure_local", &self.allow_insecure_local)
1142                .field("retry_policy", &self.retry_policy)
1143                .finish()
1144        }
1145    }
1146
1147    /// Builder for a CockroachDB-compatible SQLx PostgreSQL pool.
1148    #[derive(Clone, Debug)]
1149    pub struct CockroachPoolBuilder {
1150        config: CockroachPoolConfig,
1151        telemetry: IntegrationTelemetry,
1152    }
1153
1154    impl CockroachPoolBuilder {
1155        /// Creates a builder from an explicit CockroachDB URL.
1156        pub fn new(database_url: impl Into<String>) -> Self {
1157            Self {
1158                config: CockroachPoolConfig::new(database_url),
1159                telemetry: IntegrationTelemetry::new(),
1160            }
1161        }
1162
1163        /// Replaces the typed CockroachDB config.
1164        pub fn config(mut self, config: CockroachPoolConfig) -> Self {
1165            self.config = config;
1166            self
1167        }
1168
1169        /// Adds shared tracing, metrics, dashboard, or custom telemetry.
1170        pub fn telemetry(mut self, telemetry: IntegrationTelemetry) -> Self {
1171            self.telemetry = telemetry;
1172            self
1173        }
1174
1175        /// Instruments operations with an existing Nidus observability observer.
1176        #[cfg(feature = "observability")]
1177        pub fn observability(
1178            mut self,
1179            observer: nidus_observability::ObservabilityAdapterObserver,
1180        ) -> Self {
1181            self.telemetry = self.telemetry.observability(observer);
1182            self
1183        }
1184
1185        /// Connects after enforcing the default `verify-full` TLS policy.
1186        pub async fn connect(self) -> Result<CockroachPoolProvider> {
1187            self.config.validate()?;
1188            let started_at = Instant::now();
1189            let connect_options =
1190                sqlx::postgres::PgConnectOptions::from_str(&self.config.database_url)?;
1191            let mut pool_options =
1192                sqlx::postgres::PgPoolOptions::new().acquire_timeout(self.config.acquire_timeout);
1193            if let Some(value) = self.config.max_connections {
1194                pool_options = pool_options.max_connections(value);
1195            }
1196            if let Some(value) = self.config.min_connections {
1197                pool_options = pool_options.min_connections(value);
1198            }
1199            let pool = pool_options.connect_with(connect_options).await;
1200            self.telemetry
1201                .record(&IntegrationEvent::new(
1202                    "nidus-sqlx-cockroach",
1203                    "connect",
1204                    if pool.is_ok() {
1205                        IntegrationStatus::Success
1206                    } else {
1207                        IntegrationStatus::Failure
1208                    },
1209                    started_at.elapsed(),
1210                ))
1211                .await;
1212            Ok(CockroachPoolProvider {
1213                pool: pool?,
1214                retry_policy: self.config.retry_policy,
1215                telemetry: self.telemetry,
1216            })
1217        }
1218
1219        /// Connects and registers the provider as a Nidus singleton.
1220        pub async fn register(self, container: &mut Container) -> Result<()> {
1221            container.register_singleton(self.connect().await?)?;
1222            Ok(())
1223        }
1224    }
1225
1226    /// Future returned by a retryable CockroachDB transaction callback.
1227    pub type CockroachTransactionFuture<'connection, T> =
1228        BoxFuture<'connection, std::result::Result<T, sqlx::Error>>;
1229
1230    /// Nidus provider exposing the native SQLx PostgreSQL pool for CockroachDB.
1231    #[derive(Clone, Debug)]
1232    pub struct CockroachPoolProvider {
1233        pool: sqlx::PgPool,
1234        retry_policy: CockroachRetryPolicy,
1235        telemetry: IntegrationTelemetry,
1236    }
1237
1238    impl CockroachPoolProvider {
1239        /// Creates a CockroachDB provider builder.
1240        pub fn builder(database_url: impl Into<String>) -> CockroachPoolBuilder {
1241            CockroachPoolBuilder::new(database_url)
1242        }
1243
1244        /// Creates a provider from an existing SQLx pool and retry policy.
1245        pub fn from_pool(pool: sqlx::PgPool, retry_policy: CockroachRetryPolicy) -> Result<Self> {
1246            retry_policy.validate()?;
1247            Ok(Self {
1248                pool,
1249                retry_policy,
1250                telemetry: IntegrationTelemetry::new(),
1251            })
1252        }
1253
1254        /// Returns direct access to the native SQLx PostgreSQL pool.
1255        pub fn pool(&self) -> &sqlx::PgPool {
1256            &self.pool
1257        }
1258
1259        /// Returns the bounded transaction retry policy.
1260        pub fn retry_policy(&self) -> &CockroachRetryPolicy {
1261            &self.retry_policy
1262        }
1263
1264        /// Runs a database-only transaction and retries SQLSTATE `40001` failures.
1265        ///
1266        /// The callback may execute more than once. It must contain only
1267        /// transactional database effects; do not perform HTTP calls, publish
1268        /// messages, send email, or mutate external state inside it. SQLSTATE
1269        /// `40003` ambiguous results are never retried automatically.
1270        pub async fn transaction_with_retry<T, F>(
1271            &self,
1272            mut operation: F,
1273        ) -> CockroachTransactionResult<T>
1274        where
1275            T: Send,
1276            F: for<'connection> FnMut(
1277                &'connection mut sqlx::PgConnection,
1278            ) -> CockroachTransactionFuture<'connection, T>,
1279        {
1280            let started_at = Instant::now();
1281            let mut connection = self.pool.acquire().await?;
1282            for attempt in 1..=self.retry_policy.max_attempts {
1283                let mut transaction = connection.begin().await?;
1284                let outcome = operation(&mut transaction).await;
1285                let final_outcome: std::result::Result<(), sqlx::Error> = match outcome {
1286                    Ok(value) => match transaction.commit().await {
1287                        Ok(()) => {
1288                            self.record_transaction(true, started_at).await;
1289                            return Ok(value);
1290                        }
1291                        Err(error) => Err(error),
1292                    },
1293                    Err(error) => {
1294                        let _ = transaction.rollback().await;
1295                        Err(error)
1296                    }
1297                };
1298
1299                let error = final_outcome.unwrap_err();
1300                if !is_retryable_serialization_error(&error) {
1301                    self.record_transaction(false, started_at).await;
1302                    return Err(error.into());
1303                }
1304                if attempt == self.retry_policy.max_attempts {
1305                    self.record_transaction(false, started_at).await;
1306                    return Err(CockroachTransactionError::RetryExhausted {
1307                        attempts: attempt,
1308                        source: error,
1309                    });
1310                }
1311                tokio::time::sleep(self.retry_policy.delay_for_retry(attempt)).await;
1312            }
1313            unreachable!("validated CockroachDB retry policy always attempts at least once")
1314        }
1315
1316        /// Executes a lightweight readiness query with a safe error response.
1317        #[cfg(feature = "health")]
1318        pub async fn health_status(&self) -> nidus_http::health::HealthStatus {
1319            let started_at = Instant::now();
1320            let result = sqlx::query("SELECT 1").execute(&self.pool).await;
1321            self.telemetry
1322                .record(&IntegrationEvent::new(
1323                    "nidus-sqlx-cockroach",
1324                    "health",
1325                    if result.is_ok() {
1326                        IntegrationStatus::Success
1327                    } else {
1328                        IntegrationStatus::Failure
1329                    },
1330                    started_at.elapsed(),
1331                ))
1332                .await;
1333            if result.is_ok() {
1334                nidus_http::health::HealthStatus::up()
1335            } else {
1336                nidus_http::health::HealthStatus::down("cockroachdb readiness check failed")
1337            }
1338        }
1339
1340        /// Adds this provider as a readiness check on a health registry.
1341        #[cfg(feature = "health")]
1342        pub fn register_ready_check(
1343            self: std::sync::Arc<Self>,
1344            registry: nidus_http::health::HealthRegistry,
1345            name: impl Into<String>,
1346        ) -> nidus_http::health::HealthRegistry {
1347            registry.ready_check(name, move || {
1348                let provider = std::sync::Arc::clone(&self);
1349                async move { provider.health_status().await }
1350            })
1351        }
1352
1353        async fn record_transaction(&self, success: bool, started_at: Instant) {
1354            self.telemetry
1355                .record(&IntegrationEvent::new(
1356                    "nidus-sqlx-cockroach",
1357                    "transaction",
1358                    if success {
1359                        IntegrationStatus::Success
1360                    } else {
1361                        IntegrationStatus::Failure
1362                    },
1363                    started_at.elapsed(),
1364                ))
1365                .await;
1366        }
1367    }
1368
1369    #[async_trait::async_trait]
1370    impl nidus_core::LifecycleHook for CockroachPoolProvider {
1371        async fn on_shutdown(&self) -> nidus_core::Result<()> {
1372            self.pool.close().await;
1373            Ok(())
1374        }
1375    }
1376
1377    fn is_retryable_serialization_error(error: &sqlx::Error) -> bool {
1378        match error {
1379            sqlx::Error::Database(database) => database.code().as_deref() == Some("40001"),
1380            _ => false,
1381        }
1382    }
1383}
1384
1385#[cfg(feature = "cockroach")]
1386pub use cockroach::{
1387    CockroachPoolBuilder, CockroachPoolConfig, CockroachPoolProvider, CockroachRetryPolicy,
1388    CockroachTransactionError, CockroachTransactionFuture, CockroachTransactionResult,
1389};
nidus_sqlx/lib.rs

nidus_sqlx/
lib.rs
