1use std::fmt;
4use std::sync::Arc;
5
6#[allow(unused_imports)]
7use crate::error::{Error, Result};
8use crate::schema::{Schema, SchemaPackage, Unbound};
9use type_bridge_orm::_registry::DescriptorRegistry;
10
11#[derive(Clone, Copy, Debug, Eq, PartialEq)]
13pub enum DatabaseCreateOutcome {
14 Created,
16 AlreadyExists,
18}
19
20#[derive(Clone, Copy, Debug, Eq, PartialEq)]
22pub enum DatabaseDeleteOutcome {
23 Deleted,
25 AlreadyAbsent,
27}
28
29#[cfg(feature = "typedb")]
31#[derive(Clone, Copy, Debug, Eq, PartialEq)]
32pub enum ManagedDatabasePairState {
33 Absent,
35 StandaloneManaged,
37 OwnedPair,
39 OwnedJournalOrphan,
41}
42
43#[cfg(feature = "typedb")]
45#[derive(Clone, Copy, Debug, Eq, PartialEq)]
46pub enum ManagedDatabaseDeleteOutcome {
47 AlreadyAbsent,
49 DeletedStandaloneManaged,
51 DeletedOwnedPair,
53 DeletedOwnedJournalOrphan,
55}
56
57#[cfg(feature = "typedb")]
59pub struct ManagedDatabaseDeletionPlan {
60 inner: type_bridge_schema_migration_typedb::ManagedDatabasePairDeletionPlan,
61}
62
63#[cfg(feature = "typedb")]
64impl ManagedDatabaseDeletionPlan {
65 #[must_use]
67 pub fn inspected_state(&self) -> ManagedDatabasePairState {
68 map_pair_state(self.inner.inspected_state())
69 }
70
71 pub async fn execute(self) -> Result<ManagedDatabaseDeleteOutcome> {
73 self.inner
74 .execute()
75 .await
76 .map(map_delete_outcome)
77 .map_err(administration_error)
78 }
79
80 pub async fn execute_controlled(
82 self,
83 control: &crate::MigrationExecutionControl,
84 ) -> Result<ManagedDatabaseDeleteOutcome> {
85 self.inner
86 .execute_controlled(control)
87 .await
88 .map(map_delete_outcome)
89 .map_err(administration_error)
90 }
91}
92
93#[derive(Clone, PartialEq, Eq)]
95pub struct ConnectionOptions {
96 address: String,
97 database: String,
98 username: Option<String>,
99 password: Option<String>,
100 http_port: u16,
101 tls: bool,
102}
103
104impl fmt::Debug for ConnectionOptions {
105 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
106 formatter
107 .debug_struct("ConnectionOptions")
108 .field("address", &self.address)
109 .field("database", &self.database)
110 .field("username", &self.username)
111 .field("password", &self.password.as_ref().map(|_| "[REDACTED]"))
112 .field("http_port", &self.http_port)
113 .field("tls", &self.tls)
114 .finish()
115 }
116}
117
118impl ConnectionOptions {
119 #[must_use]
121 pub fn new(address: impl Into<String>, database: impl Into<String>) -> Self {
122 Self {
123 address: address.into(),
124 database: database.into(),
125 username: None,
126 password: None,
127 http_port: 8000,
128 tls: false,
129 }
130 }
131
132 #[must_use]
134 pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
135 self.username = Some(username.into());
136 self.password = Some(password.into());
137 self
138 }
139
140 #[must_use]
142 pub fn http_port(mut self, port: u16) -> Self {
143 self.http_port = port;
144 self
145 }
146
147 #[must_use]
149 pub fn tls(mut self, enabled: bool) -> Self {
150 self.tls = enabled;
151 self
152 }
153
154 #[must_use]
156 pub fn address(&self) -> &str {
157 &self.address
158 }
159
160 #[must_use]
162 pub fn database(&self) -> &str {
163 &self.database
164 }
165
166 #[must_use]
168 pub fn get_http_port(&self) -> u16 {
169 self.http_port
170 }
171
172 #[must_use]
174 pub fn is_tls(&self) -> bool {
175 self.tls
176 }
177}
178
179impl From<(&str, &str)> for ConnectionOptions {
180 fn from((address, database): (&str, &str)) -> Self {
181 Self::new(address, database)
182 }
183}
184
185pub struct Database<S: Schema = Unbound> {
187 inner: type_bridge_orm::Database,
188 installed_schema: Option<Arc<type_bridge_orm::InstalledRuntimeProjection>>,
189 match_registry: Option<Arc<DescriptorRegistry>>,
190 #[cfg_attr(not(feature = "typedb"), allow(dead_code))]
191 managed_scope_id: Option<type_bridge_contract::managed_scope::ManagedScopeId>,
192 marker: std::marker::PhantomData<fn() -> S>,
193}
194
195pub(crate) fn build_match_registry(
196 installed: &type_bridge_orm::InstalledRuntimeProjection,
197) -> Result<Arc<DescriptorRegistry>> {
198 installed
199 .match_registry()
200 .map(Arc::new)
201 .map_err(Error::from_orm)
202}
203
204impl Database<Unbound> {
205 #[cfg(feature = "typedb")]
207 pub async fn connect(options: impl Into<ConnectionOptions>) -> Result<Database<Unbound>> {
208 let opts = options.into();
209 let username = opts.username.as_deref().unwrap_or("admin");
210 let password = opts.password.as_deref().unwrap_or("password");
211 let orm_opts = type_bridge_orm::ConnectOptions {
212 http_port: opts.http_port,
213 tls: opts.tls,
214 ..type_bridge_orm::ConnectOptions::default()
215 };
216
217 let inner = type_bridge_orm::Database::connect_with_options(
218 &opts.address,
219 &opts.database,
220 username,
221 password,
222 orm_opts,
223 )
224 .await
225 .map_err(Error::from_orm)?;
226
227 Ok(Database {
228 inner,
229 installed_schema: None,
230 match_registry: None,
231 managed_scope_id: None,
232 marker: std::marker::PhantomData,
233 })
234 }
235
236 #[allow(dead_code)]
238 pub(crate) fn from_orm_database(inner: type_bridge_orm::Database) -> Self {
239 Self {
240 inner,
241 installed_schema: None,
242 match_registry: None,
243 managed_scope_id: None,
244 marker: std::marker::PhantomData,
245 }
246 }
247
248 pub fn with_schema<S: Schema>(self, schema: SchemaPackage<S>) -> Result<Database<S>> {
250 let (installed, authority) = schema.verify_and_install_with_authority()?;
251 let match_registry = build_match_registry(&installed)?;
252 Ok(Database::from_bound_parts(
253 self.inner,
254 installed,
255 match_registry,
256 authority.map(|authority| authority.managed_scope().id().clone()),
257 ))
258 }
259}
260
261impl<S: Schema> Database<S> {
262 pub(crate) fn from_bound_parts(
263 inner: type_bridge_orm::Database,
264 installed: Arc<type_bridge_orm::InstalledRuntimeProjection>,
265 match_registry: Arc<DescriptorRegistry>,
266 managed_scope_id: Option<type_bridge_contract::managed_scope::ManagedScopeId>,
267 ) -> Self {
268 Self {
269 inner,
270 installed_schema: Some(installed),
271 match_registry: Some(match_registry),
272 managed_scope_id,
273 marker: std::marker::PhantomData,
274 }
275 }
276
277 #[cfg(test)]
278 pub(crate) fn from_test_parts(
279 inner: type_bridge_orm::Database,
280 installed: type_bridge_orm::InstalledRuntimeProjection,
281 ) -> Self {
282 let installed = Arc::new(installed);
283 let match_registry =
284 build_match_registry(&installed).expect("test projection descriptors register");
285 Self {
286 inner,
287 installed_schema: Some(installed),
288 match_registry: Some(match_registry),
289 managed_scope_id: None,
290 marker: std::marker::PhantomData,
291 }
292 }
293 #[cfg(test)]
294 pub(crate) fn from_test_unbound_parts(inner: type_bridge_orm::Database) -> Self {
295 Self {
296 inner,
297 installed_schema: None,
298 match_registry: None,
299 managed_scope_id: None,
300 marker: std::marker::PhantomData,
301 }
302 }
303 pub fn entities<M>(&self) -> crate::entity_manager::EntityManager<'_, S, M>
305 where
306 M: crate::__codegen::EntityModel<Schema = S>,
307 {
308 crate::entity_manager::EntityManager::new(self)
309 }
310 pub fn relations<M>(&self) -> crate::relation_manager::RelationManager<'_, S, M>
312 where
313 M: crate::__codegen::RelationModel<Schema = S>,
314 {
315 crate::relation_manager::RelationManager::new(self)
316 }
317 pub async fn write(&self) -> Result<crate::transaction::WriteTransaction<'_, S>> {
322 crate::transaction::WriteTransaction::open(self).await
323 }
324 pub async fn read(&self) -> Result<crate::transaction::ReadTransaction<'_, S>> {
327 crate::transaction::ReadTransaction::open(self).await
328 }
329 #[must_use]
331 pub fn database_name(&self) -> &str {
332 self.inner.database_name()
333 }
334
335 pub async fn database_exists(&self) -> Result<bool> {
337 self.inner.database_exists().await.map_err(Error::from_orm)
338 }
339
340 #[cfg(feature = "typedb")]
342 pub async fn database_exists_controlled(
343 &self,
344 control: &crate::MigrationExecutionControl,
345 ) -> Result<bool> {
346 if let Some(scope) = &self.managed_scope_id {
347 return self
348 .pair_administrator(scope.clone())?
349 .database_exists_controlled(control)
350 .await
351 .map_err(administration_error);
352 }
353 control.check().map_err(administration_error)?;
354 self.database_exists().await
355 }
356
357 pub async fn create_database(&self) -> Result<DatabaseCreateOutcome> {
359 #[cfg(feature = "typedb")]
360 if let Some(scope) = &self.managed_scope_id {
361 return self
362 .pair_administrator(scope.clone())?
363 .create_database_outcome()
364 .await
365 .map(|outcome| match outcome {
366 type_bridge_schema_migration_typedb::ManagedDatabasePairCreateOutcome::Created => DatabaseCreateOutcome::Created,
367 type_bridge_schema_migration_typedb::ManagedDatabasePairCreateOutcome::AlreadyExists => DatabaseCreateOutcome::AlreadyExists,
368 })
369 .map_err(administration_error);
370 }
371 self.inner
372 .create_database_outcome()
373 .await
374 .map(|outcome| match outcome {
375 type_bridge_orm::session::DatabaseCreateOutcome::Created => {
376 DatabaseCreateOutcome::Created
377 }
378 type_bridge_orm::session::DatabaseCreateOutcome::AlreadyExists => {
379 DatabaseCreateOutcome::AlreadyExists
380 }
381 })
382 .map_err(Error::from_orm)
383 }
384
385 #[cfg(feature = "typedb")]
387 pub async fn create_database_controlled(
388 &self,
389 control: &crate::MigrationExecutionControl,
390 ) -> Result<DatabaseCreateOutcome> {
391 if let Some(scope) = &self.managed_scope_id {
392 return self
393 .pair_administrator(scope.clone())?
394 .create_database_outcome_controlled(control)
395 .await
396 .map(|outcome| match outcome {
397 type_bridge_schema_migration_typedb::ManagedDatabasePairCreateOutcome::Created => DatabaseCreateOutcome::Created,
398 type_bridge_schema_migration_typedb::ManagedDatabasePairCreateOutcome::AlreadyExists => DatabaseCreateOutcome::AlreadyExists,
399 })
400 .map_err(administration_error);
401 }
402 control.check().map_err(administration_error)?;
403 self.create_database().await
404 }
405
406 pub async fn delete_database(&self) -> Result<DatabaseDeleteOutcome> {
408 #[cfg(feature = "typedb")]
409 if self.managed_scope_id.is_some() {
410 return Err(Error::Database {
411 message: "managed database deletion requires plan_database_delete() and explicit plan execution"
412 .to_owned(),
413 source: None,
414 });
415 }
416 self.inner
417 .delete_database_outcome()
418 .await
419 .map(|outcome| match outcome {
420 type_bridge_orm::session::DatabaseDeleteOutcome::Deleted => {
421 DatabaseDeleteOutcome::Deleted
422 }
423 type_bridge_orm::session::DatabaseDeleteOutcome::AlreadyAbsent => {
424 DatabaseDeleteOutcome::AlreadyAbsent
425 }
426 })
427 .map_err(Error::from_orm)
428 }
429
430 #[cfg(feature = "typedb")]
432 pub async fn inspect_database_pair(&self) -> Result<ManagedDatabasePairState> {
433 let scope = self
434 .managed_scope_id
435 .clone()
436 .ok_or_else(|| Error::Database {
437 message:
438 "managed database administration requires verified generated schema authority"
439 .to_owned(),
440 source: None,
441 })?;
442 self.pair_administrator(scope)?
443 .inspect()
444 .await
445 .map(map_pair_state)
446 .map_err(administration_error)
447 }
448
449 #[cfg(feature = "typedb")]
451 pub async fn inspect_database_pair_controlled(
452 &self,
453 control: &crate::MigrationExecutionControl,
454 ) -> Result<ManagedDatabasePairState> {
455 let scope = self
456 .managed_scope_id
457 .clone()
458 .ok_or_else(|| Error::Database {
459 message:
460 "managed database administration requires verified generated schema authority"
461 .to_owned(),
462 source: None,
463 })?;
464 self.pair_administrator(scope)?
465 .inspect_controlled(control)
466 .await
467 .map(map_pair_state)
468 .map_err(administration_error)
469 }
470
471 #[cfg(feature = "typedb")]
473 pub async fn plan_database_delete(&self) -> Result<ManagedDatabaseDeletionPlan> {
474 let scope = self
475 .managed_scope_id
476 .clone()
477 .ok_or_else(|| Error::Database {
478 message: "managed database deletion requires verified generated schema authority"
479 .to_owned(),
480 source: None,
481 })?;
482 self.pair_administrator(scope)?
483 .plan_delete()
484 .await
485 .map(|inner| ManagedDatabaseDeletionPlan { inner })
486 .map_err(administration_error)
487 }
488
489 #[cfg(feature = "typedb")]
491 pub async fn plan_database_delete_controlled(
492 &self,
493 control: &crate::MigrationExecutionControl,
494 ) -> Result<ManagedDatabaseDeletionPlan> {
495 let scope = self
496 .managed_scope_id
497 .clone()
498 .ok_or_else(|| Error::Database {
499 message: "managed database deletion requires verified generated schema authority"
500 .to_owned(),
501 source: None,
502 })?;
503 self.pair_administrator(scope)?
504 .plan_delete_controlled(control)
505 .await
506 .map(|inner| ManagedDatabaseDeletionPlan { inner })
507 .map_err(administration_error)
508 }
509
510 #[cfg(feature = "typedb")]
511 fn pair_administrator(
512 &self,
513 scope: type_bridge_contract::managed_scope::ManagedScopeId,
514 ) -> Result<type_bridge_schema_migration_typedb::ManagedDatabasePairAdministrator> {
515 type_bridge_schema_migration_typedb::ManagedDatabasePairAdministrator::from_managed_database(
516 Arc::new(self.inner.clone()),
517 scope,
518 )
519 .map_err(administration_error)
520 }
521
522 pub fn close(&self) -> Result<()> {
527 self.inner.close().map_err(Error::from_orm)
528 }
529
530 #[must_use]
532 pub fn is_schema_bound(&self) -> bool {
533 self.installed_schema.is_some()
534 }
535
536 #[allow(dead_code)]
538 pub(crate) fn inner_orm(&self) -> &type_bridge_orm::Database {
539 &self.inner
540 }
541
542 pub(crate) fn operation_limits(
543 &self,
544 requested: type_bridge_orm::QueryExecutionResourceLimits,
545 ) -> type_bridge_orm::QueryExecutionResourceLimits {
546 self.inner.answer_limits().map_or_else(
547 || requested.effective(),
548 |ceiling| requested.constrained_by(ceiling),
549 )
550 }
551
552 #[allow(dead_code)]
554 pub(crate) fn installed_schema(
555 &self,
556 ) -> Option<&Arc<type_bridge_orm::InstalledRuntimeProjection>> {
557 self.installed_schema.as_ref()
558 }
559
560 #[allow(dead_code)]
562 pub(crate) fn match_registry(&self) -> Option<&Arc<DescriptorRegistry>> {
563 self.match_registry.as_ref()
564 }
565}
566
567#[cfg(feature = "typedb")]
568fn map_pair_state(
569 state: type_bridge_schema_migration_typedb::ManagedDatabasePairState,
570) -> ManagedDatabasePairState {
571 match state {
572 type_bridge_schema_migration_typedb::ManagedDatabasePairState::Absent => {
573 ManagedDatabasePairState::Absent
574 }
575 type_bridge_schema_migration_typedb::ManagedDatabasePairState::StandaloneManaged => {
576 ManagedDatabasePairState::StandaloneManaged
577 }
578 type_bridge_schema_migration_typedb::ManagedDatabasePairState::OwnedPair => {
579 ManagedDatabasePairState::OwnedPair
580 }
581 type_bridge_schema_migration_typedb::ManagedDatabasePairState::OwnedJournalOrphan => {
582 ManagedDatabasePairState::OwnedJournalOrphan
583 }
584 }
585}
586
587#[cfg(feature = "typedb")]
588fn map_delete_outcome(
589 outcome: type_bridge_schema_migration_typedb::ManagedDatabasePairDeleteOutcome,
590) -> ManagedDatabaseDeleteOutcome {
591 match outcome {
592 type_bridge_schema_migration_typedb::ManagedDatabasePairDeleteOutcome::AlreadyAbsent => {
593 ManagedDatabaseDeleteOutcome::AlreadyAbsent
594 }
595 type_bridge_schema_migration_typedb::ManagedDatabasePairDeleteOutcome::DeletedStandaloneManaged => ManagedDatabaseDeleteOutcome::DeletedStandaloneManaged,
596 type_bridge_schema_migration_typedb::ManagedDatabasePairDeleteOutcome::DeletedOwnedPair => {
597 ManagedDatabaseDeleteOutcome::DeletedOwnedPair
598 }
599 type_bridge_schema_migration_typedb::ManagedDatabasePairDeleteOutcome::DeletedOwnedJournalOrphan => ManagedDatabaseDeleteOutcome::DeletedOwnedJournalOrphan,
600 }
601}
602
603#[cfg(feature = "typedb")]
604fn administration_error(error: type_bridge_contract::diagnostic::Diagnostic) -> Error {
605 Error::from_contract_diagnostic(error)
606}
607
608#[cfg(test)]
609mod tests {
610 use std::collections::BTreeSet;
611 use std::sync::Arc;
612 use std::sync::Mutex;
613 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
614
615 use type_bridge_orm::error::OrmError;
616 use type_bridge_orm::session::backend::{BoxFuture, DriverBackend, TransactionOps, TxType};
617
618 use super::Database;
619 use crate::schema::Unbound;
620
621 struct CloseBackend {
622 closed: Arc<AtomicBool>,
623 close_calls: Arc<AtomicUsize>,
624 }
625
626 struct AdministrationBackend {
627 databases: Arc<Mutex<BTreeSet<String>>>,
628 }
629
630 impl DriverBackend for AdministrationBackend {
631 fn open_transaction(
632 &self,
633 _database: &str,
634 _tx_type: TxType,
635 ) -> BoxFuture<'_, std::result::Result<Box<dyn TransactionOps>, OrmError>> {
636 Box::pin(async { Err(OrmError::Connection("unexpected transaction".into())) })
637 }
638
639 fn is_open(&self) -> bool {
640 true
641 }
642
643 fn database_exists(
644 &self,
645 database: &str,
646 ) -> BoxFuture<'_, std::result::Result<bool, OrmError>> {
647 let exists = self.databases.lock().unwrap().contains(database);
648 Box::pin(async move { Ok(exists) })
649 }
650
651 fn create_database(
652 &self,
653 database: &str,
654 ) -> BoxFuture<'_, std::result::Result<(), OrmError>> {
655 self.databases.lock().unwrap().insert(database.to_owned());
656 Box::pin(async { Ok(()) })
657 }
658
659 fn delete_database(
660 &self,
661 database: &str,
662 ) -> BoxFuture<'_, std::result::Result<(), OrmError>> {
663 self.databases.lock().unwrap().remove(database);
664 Box::pin(async { Ok(()) })
665 }
666
667 fn schema_text(
668 &self,
669 _database: &str,
670 ) -> BoxFuture<'_, std::result::Result<String, OrmError>> {
671 Box::pin(async { Ok(String::new()) })
672 }
673 }
674
675 impl DriverBackend for CloseBackend {
676 fn open_transaction(
677 &self,
678 _database: &str,
679 _tx_type: TxType,
680 ) -> BoxFuture<'_, std::result::Result<Box<dyn TransactionOps>, OrmError>> {
681 Box::pin(async {
682 Err(OrmError::Connection(
683 "closed test backend cannot open transactions".into(),
684 ))
685 })
686 }
687
688 fn is_open(&self) -> bool {
689 !self.closed.load(Ordering::SeqCst)
690 }
691
692 fn close_connection(&self) -> std::result::Result<(), OrmError> {
693 self.close_calls.fetch_add(1, Ordering::SeqCst);
694 self.closed.store(true, Ordering::SeqCst);
695 Ok(())
696 }
697 }
698
699 #[test]
700 fn explicit_database_close_is_idempotent() {
701 let closed = Arc::new(AtomicBool::new(false));
702 let close_calls = Arc::new(AtomicUsize::new(0));
703 let inner = type_bridge_orm::Database::with_backend(
704 Box::new(CloseBackend {
705 closed: Arc::clone(&closed),
706 close_calls: Arc::clone(&close_calls),
707 }),
708 "app",
709 );
710 let database: Database<Unbound> = Database::from_test_unbound_parts(inner);
711
712 database.close().unwrap();
713 database.close().unwrap();
714
715 assert!(closed.load(Ordering::SeqCst));
716 assert_eq!(close_calls.load(Ordering::SeqCst), 2);
717 }
718
719 #[cfg(feature = "typedb")]
720 #[tokio::test]
721 async fn generated_database_administration_is_pair_aware_and_plan_owned() {
722 let databases = Arc::new(Mutex::new(BTreeSet::new()));
723 let inner = type_bridge_orm::Database::with_backend(
724 Box::new(AdministrationBackend {
725 databases: Arc::clone(&databases),
726 }),
727 "app",
728 );
729 let database: Database<Unbound> = Database {
730 inner,
731 installed_schema: None,
732 match_registry: None,
733 managed_scope_id: Some(
734 type_bridge_contract::managed_scope::ManagedScopeId::new("generated-scope")
735 .unwrap(),
736 ),
737 marker: std::marker::PhantomData,
738 };
739
740 let cancellation = type_bridge_schema_migration::MigrationCancellation::default();
741 cancellation.cancel();
742 let control = type_bridge_schema_migration::MigrationExecutionControl::new(
743 cancellation,
744 None,
745 type_bridge_schema_migration::MigrationExecutionResourceLimits::default(),
746 );
747 let cancelled = database
748 .database_exists_controlled(&control)
749 .await
750 .expect_err("pre-cancelled administration rejects before an effect");
751 assert_eq!(cancelled.code(), Some("migration_execution_cancelled"));
752 assert_eq!(cancelled.category(), crate::ErrorCategory::Cancelled);
753
754 assert_eq!(
755 database.create_database().await.unwrap(),
756 super::DatabaseCreateOutcome::Created
757 );
758 assert_eq!(
759 database.inspect_database_pair().await.unwrap(),
760 super::ManagedDatabasePairState::StandaloneManaged
761 );
762 assert!(database.delete_database().await.is_err());
763 let plan = database.plan_database_delete().await.unwrap();
764 drop(database);
765 assert_eq!(
766 plan.execute().await.unwrap(),
767 super::ManagedDatabaseDeleteOutcome::DeletedStandaloneManaged
768 );
769 assert!(databases.lock().unwrap().is_empty());
770 }
771}