1use std::collections::BTreeSet;
4use std::marker::PhantomData;
5use std::sync::Arc;
6
7use type_bridge_contract::fingerprint::Fingerprint;
8use type_bridge_contract::migration::{MigrationId, MigrationStep};
9use type_bridge_schema::{ManagedDeltaContext, SafetyClass};
10
11use crate::error::{Error, Result};
12use crate::schema::{Schema, SchemaPackage};
13#[cfg(feature = "typedb")]
14use crate::session::Database;
15
16#[derive(Clone, Debug)]
18pub struct MigrationCatalog<S: Schema> {
19 inner: type_bridge_schema_migration::MigrationCatalog,
20 marker: PhantomData<fn() -> S>,
21}
22
23impl<S: Schema> MigrationCatalog<S> {
24 pub fn identity(&self, app_label: &str, name: &str) -> Result<MigrationId> {
26 let _ = self;
27 MigrationId::new(app_label, name).map_err(migration_diagnostic)
28 }
29
30 #[must_use]
32 pub const fn fingerprint(&self) -> &Fingerprint {
33 self.inner.fingerprint()
34 }
35
36 #[must_use]
38 pub fn len(&self) -> usize {
39 self.inner.entries().len()
40 }
41
42 #[must_use]
44 pub fn is_empty(&self) -> bool {
45 self.inner.entries().is_empty()
46 }
47
48 #[must_use]
50 pub fn heads(&self) -> &[MigrationId] {
51 self.inner.heads()
52 }
53
54 #[must_use]
56 pub fn entry(&self, index: usize) -> Option<MigrationHistoryEntry<'_>> {
57 self.inner
58 .entries()
59 .get(index)
60 .map(|entry| MigrationHistoryEntry { inner: entry })
61 }
62
63 pub fn preview_apply(
65 &self,
66 applied: impl IntoIterator<Item = MigrationId>,
67 targets: Option<Vec<MigrationId>>,
68 ) -> Result<MigrationPreview<S>> {
69 let applied = applied.into_iter().collect();
70 let target = match targets {
71 Some(targets) => type_bridge_schema_migration::MigrationApplyTarget::Explicit(
72 targets.into_iter().collect(),
73 ),
74 None => type_bridge_schema_migration::MigrationApplyTarget::DefaultHead,
75 };
76 self.inner
77 .preview_apply(&applied, &target)
78 .map(|plan| MigrationPreview {
79 inner: Arc::new(MigrationPreviewState {
80 catalog: self.inner.clone(),
81 plan: MigrationPreviewInner::Apply(plan),
82 }),
83 marker: PhantomData,
84 })
85 .map_err(plan_error)
86 }
87
88 pub fn preview_rollback(
90 &self,
91 applied: impl IntoIterator<Item = MigrationId>,
92 removals: impl IntoIterator<Item = MigrationId>,
93 ) -> Result<MigrationPreview<S>> {
94 self.inner
95 .preview_rollback(
96 &applied.into_iter().collect(),
97 &removals.into_iter().collect(),
98 )
99 .map(|plan| MigrationPreview {
100 inner: Arc::new(MigrationPreviewState {
101 catalog: self.inner.clone(),
102 plan: MigrationPreviewInner::Rollback(plan),
103 }),
104 marker: PhantomData,
105 })
106 .map_err(plan_error)
107 }
108
109 #[cfg(feature = "typedb")]
111 pub async fn verify(
112 &self,
113 database: &Database<S>,
114 ) -> Result<type_bridge_schema_migration::MigrationVerifyReport> {
115 type_bridge_schema_migration_typedb::verify_catalog_state(
116 Arc::new(database.inner_orm().clone()),
117 &self.inner,
118 )
119 .await
120 .map_err(migration_diagnostic)
121 }
122
123 #[cfg(feature = "typedb")]
125 pub async fn applied_migrations(&self, database: &Database<S>) -> Result<Vec<MigrationId>> {
126 type_bridge_schema_migration_typedb::load_catalog_applied_migrations(
127 Arc::new(database.inner_orm().clone()),
128 &self.inner,
129 )
130 .await
131 .map_err(migration_diagnostic)
132 }
133}
134
135#[derive(Clone, Debug)]
136#[allow(clippy::large_enum_variant)]
137enum MigrationPreviewInner {
138 Apply(type_bridge_schema_migration::VerifiedMigrationApplyPlan),
139 Rollback(type_bridge_schema_migration::VerifiedMigrationRollbackPlan),
140}
141
142#[derive(Clone, Debug)]
143struct MigrationPreviewState {
144 catalog: type_bridge_schema_migration::MigrationCatalog,
145 plan: MigrationPreviewInner,
146}
147
148#[derive(Clone, Debug)]
150pub struct MigrationPreview<S: Schema> {
151 inner: Arc<MigrationPreviewState>,
152 marker: PhantomData<fn() -> S>,
153}
154
155impl<S: Schema> MigrationPreview<S> {
156 #[must_use]
158 pub fn is_apply(&self) -> bool {
159 matches!(self.inner.plan, MigrationPreviewInner::Apply(_))
160 }
161
162 #[must_use]
164 pub fn execution_authorized(&self) -> bool {
165 match &self.inner.plan {
166 MigrationPreviewInner::Apply(plan) => plan.execution_authorized(),
167 MigrationPreviewInner::Rollback(plan) => plan.execution_authorized(),
168 }
169 }
170
171 #[must_use]
173 pub fn len(&self) -> usize {
174 match &self.inner.plan {
175 MigrationPreviewInner::Apply(plan) => plan.migrations().len(),
176 MigrationPreviewInner::Rollback(plan) => plan.rollbacks().len(),
177 }
178 }
179
180 #[must_use]
182 pub fn is_empty(&self) -> bool {
183 self.len() == 0
184 }
185
186 #[must_use]
188 pub fn entry(&self, index: usize) -> Option<MigrationPreviewEntry<'_>> {
189 match &self.inner.plan {
190 MigrationPreviewInner::Apply(plan) => {
191 plan.migrations()
192 .get(index)
193 .map(|entry| MigrationPreviewEntry {
194 id: entry.manifest().id(),
195 safety: entry.manifest().safety(),
196 step_count: entry.steps().len(),
197 transaction_group_count: entry.transaction_groups().len(),
198 backfill_count: entry.backfill_step_indices().len(),
199 reversible: entry.manifest().reversible(),
200 })
201 }
202 MigrationPreviewInner::Rollback(plan) => {
203 plan.rollbacks()
204 .get(index)
205 .map(|entry| MigrationPreviewEntry {
206 id: entry.manifest().id(),
207 safety: entry.rollback_safety(),
208 step_count: entry.operations().len(),
209 transaction_group_count: entry.steps().len(),
210 backfill_count: entry.backfills().len(),
211 reversible: true,
212 })
213 }
214 }
215 }
216
217 #[must_use]
219 pub fn approval_builder(&self) -> MigrationApprovalBuilder<S> {
220 MigrationApprovalBuilder {
221 preview: Arc::clone(&self.inner),
222 selected: BTreeSet::new(),
223 marker: PhantomData,
224 }
225 }
226
227 pub fn authorize(&self, approvals: &MigrationApprovalSet<S>) -> Result<MigrationPlan<S>> {
229 if !Arc::ptr_eq(&self.inner, &approvals.preview) {
230 return Err(migration_error("migration_approval_plan_mismatch"));
231 }
232 let policy = type_bridge_schema_migration::MigrationSafetyPolicy::default_policy();
233 let plan = match &self.inner.plan {
234 MigrationPreviewInner::Apply(plan) => self
235 .inner
236 .catalog
237 .authorize_apply(
238 &plan.applied_migrations().iter().cloned().collect(),
239 &type_bridge_schema_migration::MigrationApplyTarget::Explicit(
240 plan.target_frontier().iter().cloned().collect(),
241 ),
242 &policy,
243 &approvals.approvals,
244 )
245 .map(MigrationPreviewInner::Apply),
246 MigrationPreviewInner::Rollback(plan) => self
247 .inner
248 .catalog
249 .authorize_rollback(
250 &plan.applied_basis(),
251 &plan
252 .rollbacks()
253 .iter()
254 .map(|entry| entry.manifest().id().clone())
255 .collect(),
256 &policy,
257 &approvals.approvals,
258 )
259 .map(MigrationPreviewInner::Rollback),
260 }
261 .map_err(plan_error)?;
262 Ok(MigrationPlan {
263 catalog: self.inner.catalog.clone(),
264 plan,
265 marker: PhantomData,
266 })
267 }
268}
269
270#[derive(Debug)]
272pub struct MigrationApprovalBuilder<S: Schema> {
273 preview: Arc<MigrationPreviewState>,
274 selected: BTreeSet<usize>,
275 marker: PhantomData<fn() -> S>,
276}
277
278impl<S: Schema> MigrationApprovalBuilder<S> {
279 pub fn approve(&mut self, index: usize) -> Result<()> {
281 let safety = preview_entry(&self.preview.plan, index)
282 .ok_or_else(|| migration_error("migration_approval_index_invalid"))?
283 .safety();
284 match type_bridge_schema_migration::MigrationSafetyPolicy::default_policy().decision(safety)
285 {
286 type_bridge_schema_migration::SafetyPolicyDecision::RequireApproval => {
287 self.selected.insert(index);
288 Ok(())
289 }
290 type_bridge_schema_migration::SafetyPolicyDecision::Allow => {
291 Err(migration_error("migration_approval_not_required"))
292 }
293 type_bridge_schema_migration::SafetyPolicyDecision::Reject => {
294 Err(migration_error("migration_approval_policy_rejected"))
295 }
296 }
297 }
298
299 pub fn finish(self) -> Result<MigrationApprovalSet<S>> {
301 let approvals = self
302 .selected
303 .iter()
304 .map(|index| match &self.preview.plan {
305 MigrationPreviewInner::Apply(plan) => {
306 type_bridge_schema_migration::MigrationApplyApproval::for_manifest(
307 plan.migrations()[*index].manifest(),
308 )
309 }
310 MigrationPreviewInner::Rollback(plan) => {
311 type_bridge_schema_migration::MigrationApplyApproval::for_rollback(
312 plan.rollbacks()[*index].manifest(),
313 plan.rollbacks()[*index].rollback_safety(),
314 )
315 }
316 })
317 .collect::<std::result::Result<Vec<_>, _>>()
318 .map_err(migration_diagnostic)?;
319 Ok(MigrationApprovalSet {
320 preview: self.preview,
321 approvals,
322 marker: PhantomData,
323 })
324 }
325}
326
327#[derive(Clone, Debug)]
329pub struct MigrationApprovalSet<S: Schema> {
330 preview: Arc<MigrationPreviewState>,
331 approvals: Vec<type_bridge_schema_migration::MigrationApplyApproval>,
332 marker: PhantomData<fn() -> S>,
333}
334
335impl<S: Schema> MigrationApprovalSet<S> {
336 #[must_use]
338 pub fn len(&self) -> usize {
339 self.approvals.len()
340 }
341
342 #[must_use]
344 pub fn is_empty(&self) -> bool {
345 self.approvals.is_empty()
346 }
347}
348
349#[derive(Clone, Debug)]
351pub struct MigrationPlan<S: Schema> {
352 #[cfg_attr(not(feature = "typedb"), allow(dead_code))]
353 catalog: type_bridge_schema_migration::MigrationCatalog,
354 plan: MigrationPreviewInner,
355 marker: PhantomData<fn() -> S>,
356}
357
358impl<S: Schema> MigrationPlan<S> {
359 #[must_use]
361 pub fn execution_authorized(&self) -> bool {
362 match &self.plan {
363 MigrationPreviewInner::Apply(plan) => plan.execution_authorized(),
364 MigrationPreviewInner::Rollback(plan) => plan.execution_authorized(),
365 }
366 }
367
368 #[cfg(feature = "typedb")]
370 pub async fn execute(
371 &self,
372 database: &Database<S>,
373 holder: &str,
374 ) -> Result<type_bridge_schema_migration::MigrationExecutionReport> {
375 self.execute_controlled(
376 database,
377 holder,
378 &type_bridge_schema_migration::MigrationExecutionControl::default(),
379 )
380 .await
381 }
382
383 #[cfg(feature = "typedb")]
385 pub async fn execute_controlled(
386 &self,
387 database: &Database<S>,
388 holder: &str,
389 control: &type_bridge_schema_migration::MigrationExecutionControl,
390 ) -> Result<type_bridge_schema_migration::MigrationExecutionReport> {
391 let holder = type_bridge_schema_migration::LeaseHolderId::new(holder)
392 .map_err(migration_diagnostic)?;
393 let database = Arc::new(database.inner_orm().clone());
394 match &self.plan {
395 MigrationPreviewInner::Apply(plan) => {
396 type_bridge_schema_migration_typedb::execute_catalog_apply_plan_controlled(
397 database,
398 &self.catalog,
399 &holder,
400 plan,
401 control,
402 )
403 .await
404 .map(type_bridge_schema_migration::MigrationExecutionReport::from_apply)
405 .map_err(migration_diagnostic)
406 }
407 MigrationPreviewInner::Rollback(plan) => {
408 type_bridge_schema_migration_typedb::execute_catalog_rollback_plan_controlled(
409 database,
410 &self.catalog,
411 &holder,
412 plan,
413 control,
414 )
415 .await
416 .map(type_bridge_schema_migration::MigrationExecutionReport::from_rollback)
417 .map_err(migration_diagnostic)
418 }
419 }
420 }
421}
422
423#[derive(Clone, Copy, Debug)]
425pub struct MigrationPreviewEntry<'a> {
426 id: &'a MigrationId,
427 safety: SafetyClass,
428 step_count: usize,
429 transaction_group_count: usize,
430 backfill_count: usize,
431 reversible: bool,
432}
433
434impl MigrationPreviewEntry<'_> {
435 pub const fn id(&self) -> &MigrationId {
437 self.id
438 }
439 pub const fn safety(&self) -> SafetyClass {
441 self.safety
442 }
443 pub const fn step_count(&self) -> usize {
445 self.step_count
446 }
447 pub const fn transaction_group_count(&self) -> usize {
449 self.transaction_group_count
450 }
451 pub const fn backfill_count(&self) -> usize {
453 self.backfill_count
454 }
455 pub const fn is_reversible(&self) -> bool {
457 self.reversible
458 }
459}
460
461fn preview_entry(plan: &MigrationPreviewInner, index: usize) -> Option<MigrationPreviewEntry<'_>> {
462 match plan {
463 MigrationPreviewInner::Apply(plan) => {
464 plan.migrations()
465 .get(index)
466 .map(|entry| MigrationPreviewEntry {
467 id: entry.manifest().id(),
468 safety: entry.manifest().safety(),
469 step_count: entry.steps().len(),
470 transaction_group_count: entry.transaction_groups().len(),
471 backfill_count: entry.backfill_step_indices().len(),
472 reversible: entry.manifest().reversible(),
473 })
474 }
475 MigrationPreviewInner::Rollback(plan) => {
476 plan.rollbacks()
477 .get(index)
478 .map(|entry| MigrationPreviewEntry {
479 id: entry.manifest().id(),
480 safety: entry.rollback_safety(),
481 step_count: entry.operations().len(),
482 transaction_group_count: entry.steps().len(),
483 backfill_count: entry.backfills().len(),
484 reversible: true,
485 })
486 }
487 }
488}
489
490#[derive(Clone, Copy, Debug)]
492pub struct MigrationHistoryEntry<'a> {
493 inner: &'a type_bridge_schema_migration::VerifiedMigrationHistoryBundleEntry,
494}
495
496impl MigrationHistoryEntry<'_> {
497 #[must_use]
499 pub const fn id(&self) -> &MigrationId {
500 self.inner.manifest().id()
501 }
502
503 #[must_use]
505 pub fn parents(&self) -> &[MigrationId] {
506 self.inner.manifest().parents()
507 }
508
509 #[must_use]
511 pub const fn manifest_digest(
512 &self,
513 ) -> type_bridge_contract::migration::MigrationManifestDigest {
514 self.inner.manifest_digest()
515 }
516
517 #[must_use]
519 pub fn step_count(&self) -> usize {
520 self.inner.manifest().steps().len()
521 }
522
523 #[must_use]
525 pub fn step(&self, index: usize) -> Option<&MigrationStep> {
526 self.inner.manifest().steps().get(index)
527 }
528
529 #[must_use]
531 pub const fn safety(&self) -> SafetyClass {
532 self.inner.manifest().safety()
533 }
534
535 #[must_use]
537 pub const fn is_reversible(&self) -> bool {
538 self.inner.manifest().reversible()
539 }
540}
541
542impl<S: Schema> SchemaPackage<S> {
543 pub fn open_migration_catalog(&self, bytes: &[u8]) -> Result<MigrationCatalog<S>> {
549 let (_projection, authority) = self.verify_and_install_with_authority()?;
550 let authority = authority.ok_or_else(|| Error::SchemaVerification {
551 message: "migration catalogs require generated schema authority".to_owned(),
552 source: None,
553 })?;
554 let context = ManagedDeltaContext::new(
555 authority.managed_scope().id().clone(),
556 authority.semantic_profile().id().clone(),
557 type_bridge_schema_migration::migration_runtime_capability_vocabulary().map_err(
558 |error| Error::SchemaVerification {
559 message: format!(
560 "migration runtime capability vocabulary is invalid [{}]",
561 error.code().as_str()
562 ),
563 source: Some(Box::new(error)),
564 },
565 )?,
566 );
567 let inner = type_bridge_schema_migration::MigrationCatalog::open(bytes, &context).map_err(
568 |error| Error::SchemaVerification {
569 message: format!(
570 "generated migration catalog was rejected [{}]",
571 error.code().as_str()
572 ),
573 source: Some(Box::new(error)),
574 },
575 )?;
576 Ok(MigrationCatalog {
577 inner,
578 marker: PhantomData,
579 })
580 }
581}
582
583fn plan_error(error: type_bridge_schema_migration::MigrationApplyPlanError) -> Error {
584 let error = match error {
585 type_bridge_schema_migration::MigrationApplyPlanError::Contract(diagnostic) => {
586 return Error::from_contract_diagnostic(diagnostic);
587 }
588 error => error,
589 };
590 let message = match &error {
591 type_bridge_schema_migration::MigrationApplyPlanError::Contract(_) => unreachable!(),
592 type_bridge_schema_migration::MigrationApplyPlanError::Schema(_) => {
593 "generated migration preview schema replay failed".to_owned()
594 }
595 type_bridge_schema_migration::MigrationApplyPlanError::Lowering(diagnostic) => format!(
596 "generated migration preview lowering was rejected [{}]",
597 diagnostic.code()
598 ),
599 };
600 Error::SchemaVerification {
601 message,
602 source: Some(Box::new(error)),
603 }
604}
605
606fn migration_error(code: &str) -> Error {
607 Error::SchemaVerification {
608 message: format!("generated migration operation was rejected [{code}]"),
609 source: None,
610 }
611}
612
613fn migration_diagnostic(error: type_bridge_contract::diagnostic::Diagnostic) -> Error {
614 Error::from_contract_diagnostic(error)
615}