use std::collections::BTreeSet;
use std::marker::PhantomData;
use std::sync::Arc;
use type_bridge_contract::fingerprint::Fingerprint;
use type_bridge_contract::migration::{MigrationId, MigrationStep};
use type_bridge_schema::{ManagedDeltaContext, SafetyClass};
use crate::error::{Error, Result};
use crate::schema::{Schema, SchemaPackage};
#[cfg(feature = "typedb")]
use crate::session::Database;
#[derive(Clone, Debug)]
pub struct MigrationCatalog<S: Schema> {
inner: type_bridge_schema_migration::MigrationCatalog,
marker: PhantomData<fn() -> S>,
}
impl<S: Schema> MigrationCatalog<S> {
pub fn identity(&self, app_label: &str, name: &str) -> Result<MigrationId> {
let _ = self;
MigrationId::new(app_label, name).map_err(migration_diagnostic)
}
#[must_use]
pub const fn fingerprint(&self) -> &Fingerprint {
self.inner.fingerprint()
}
#[must_use]
pub fn len(&self) -> usize {
self.inner.entries().len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.inner.entries().is_empty()
}
#[must_use]
pub fn heads(&self) -> &[MigrationId] {
self.inner.heads()
}
#[must_use]
pub fn entry(&self, index: usize) -> Option<MigrationHistoryEntry<'_>> {
self.inner
.entries()
.get(index)
.map(|entry| MigrationHistoryEntry { inner: entry })
}
pub fn preview_apply(
&self,
applied: impl IntoIterator<Item = MigrationId>,
targets: Option<Vec<MigrationId>>,
) -> Result<MigrationPreview<S>> {
let applied = applied.into_iter().collect();
let target = match targets {
Some(targets) => type_bridge_schema_migration::MigrationApplyTarget::Explicit(
targets.into_iter().collect(),
),
None => type_bridge_schema_migration::MigrationApplyTarget::DefaultHead,
};
self.inner
.preview_apply(&applied, &target)
.map(|plan| MigrationPreview {
inner: Arc::new(MigrationPreviewState {
catalog: self.inner.clone(),
plan: MigrationPreviewInner::Apply(plan),
}),
marker: PhantomData,
})
.map_err(plan_error)
}
pub fn preview_rollback(
&self,
applied: impl IntoIterator<Item = MigrationId>,
removals: impl IntoIterator<Item = MigrationId>,
) -> Result<MigrationPreview<S>> {
self.inner
.preview_rollback(
&applied.into_iter().collect(),
&removals.into_iter().collect(),
)
.map(|plan| MigrationPreview {
inner: Arc::new(MigrationPreviewState {
catalog: self.inner.clone(),
plan: MigrationPreviewInner::Rollback(plan),
}),
marker: PhantomData,
})
.map_err(plan_error)
}
#[cfg(feature = "typedb")]
pub async fn verify(
&self,
database: &Database<S>,
) -> Result<type_bridge_schema_migration::MigrationVerifyReport> {
type_bridge_schema_migration_typedb::verify_catalog_state(
Arc::new(database.inner_orm().clone()),
&self.inner,
)
.await
.map_err(migration_diagnostic)
}
#[cfg(feature = "typedb")]
pub async fn applied_migrations(&self, database: &Database<S>) -> Result<Vec<MigrationId>> {
type_bridge_schema_migration_typedb::load_catalog_applied_migrations(
Arc::new(database.inner_orm().clone()),
&self.inner,
)
.await
.map_err(migration_diagnostic)
}
}
#[derive(Clone, Debug)]
#[allow(clippy::large_enum_variant)]
enum MigrationPreviewInner {
Apply(type_bridge_schema_migration::VerifiedMigrationApplyPlan),
Rollback(type_bridge_schema_migration::VerifiedMigrationRollbackPlan),
}
#[derive(Clone, Debug)]
struct MigrationPreviewState {
catalog: type_bridge_schema_migration::MigrationCatalog,
plan: MigrationPreviewInner,
}
#[derive(Clone, Debug)]
pub struct MigrationPreview<S: Schema> {
inner: Arc<MigrationPreviewState>,
marker: PhantomData<fn() -> S>,
}
impl<S: Schema> MigrationPreview<S> {
#[must_use]
pub fn is_apply(&self) -> bool {
matches!(self.inner.plan, MigrationPreviewInner::Apply(_))
}
#[must_use]
pub fn execution_authorized(&self) -> bool {
match &self.inner.plan {
MigrationPreviewInner::Apply(plan) => plan.execution_authorized(),
MigrationPreviewInner::Rollback(plan) => plan.execution_authorized(),
}
}
#[must_use]
pub fn len(&self) -> usize {
match &self.inner.plan {
MigrationPreviewInner::Apply(plan) => plan.migrations().len(),
MigrationPreviewInner::Rollback(plan) => plan.rollbacks().len(),
}
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.len() == 0
}
#[must_use]
pub fn entry(&self, index: usize) -> Option<MigrationPreviewEntry<'_>> {
match &self.inner.plan {
MigrationPreviewInner::Apply(plan) => {
plan.migrations()
.get(index)
.map(|entry| MigrationPreviewEntry {
id: entry.manifest().id(),
safety: entry.manifest().safety(),
step_count: entry.steps().len(),
transaction_group_count: entry.transaction_groups().len(),
backfill_count: entry.backfill_step_indices().len(),
reversible: entry.manifest().reversible(),
})
}
MigrationPreviewInner::Rollback(plan) => {
plan.rollbacks()
.get(index)
.map(|entry| MigrationPreviewEntry {
id: entry.manifest().id(),
safety: entry.rollback_safety(),
step_count: entry.operations().len(),
transaction_group_count: entry.steps().len(),
backfill_count: entry.backfills().len(),
reversible: true,
})
}
}
}
#[must_use]
pub fn approval_builder(&self) -> MigrationApprovalBuilder<S> {
MigrationApprovalBuilder {
preview: Arc::clone(&self.inner),
selected: BTreeSet::new(),
marker: PhantomData,
}
}
pub fn authorize(&self, approvals: &MigrationApprovalSet<S>) -> Result<MigrationPlan<S>> {
if !Arc::ptr_eq(&self.inner, &approvals.preview) {
return Err(migration_error("migration_approval_plan_mismatch"));
}
let policy = type_bridge_schema_migration::MigrationSafetyPolicy::default_policy();
let plan = match &self.inner.plan {
MigrationPreviewInner::Apply(plan) => self
.inner
.catalog
.authorize_apply(
&plan.applied_migrations().iter().cloned().collect(),
&type_bridge_schema_migration::MigrationApplyTarget::Explicit(
plan.target_frontier().iter().cloned().collect(),
),
&policy,
&approvals.approvals,
)
.map(MigrationPreviewInner::Apply),
MigrationPreviewInner::Rollback(plan) => self
.inner
.catalog
.authorize_rollback(
&plan.applied_basis(),
&plan
.rollbacks()
.iter()
.map(|entry| entry.manifest().id().clone())
.collect(),
&policy,
&approvals.approvals,
)
.map(MigrationPreviewInner::Rollback),
}
.map_err(plan_error)?;
Ok(MigrationPlan {
catalog: self.inner.catalog.clone(),
plan,
marker: PhantomData,
})
}
}
#[derive(Debug)]
pub struct MigrationApprovalBuilder<S: Schema> {
preview: Arc<MigrationPreviewState>,
selected: BTreeSet<usize>,
marker: PhantomData<fn() -> S>,
}
impl<S: Schema> MigrationApprovalBuilder<S> {
pub fn approve(&mut self, index: usize) -> Result<()> {
let safety = preview_entry(&self.preview.plan, index)
.ok_or_else(|| migration_error("migration_approval_index_invalid"))?
.safety();
match type_bridge_schema_migration::MigrationSafetyPolicy::default_policy().decision(safety)
{
type_bridge_schema_migration::SafetyPolicyDecision::RequireApproval => {
self.selected.insert(index);
Ok(())
}
type_bridge_schema_migration::SafetyPolicyDecision::Allow => {
Err(migration_error("migration_approval_not_required"))
}
type_bridge_schema_migration::SafetyPolicyDecision::Reject => {
Err(migration_error("migration_approval_policy_rejected"))
}
}
}
pub fn finish(self) -> Result<MigrationApprovalSet<S>> {
let approvals = self
.selected
.iter()
.map(|index| match &self.preview.plan {
MigrationPreviewInner::Apply(plan) => {
type_bridge_schema_migration::MigrationApplyApproval::for_manifest(
plan.migrations()[*index].manifest(),
)
}
MigrationPreviewInner::Rollback(plan) => {
type_bridge_schema_migration::MigrationApplyApproval::for_rollback(
plan.rollbacks()[*index].manifest(),
plan.rollbacks()[*index].rollback_safety(),
)
}
})
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(migration_diagnostic)?;
Ok(MigrationApprovalSet {
preview: self.preview,
approvals,
marker: PhantomData,
})
}
}
#[derive(Clone, Debug)]
pub struct MigrationApprovalSet<S: Schema> {
preview: Arc<MigrationPreviewState>,
approvals: Vec<type_bridge_schema_migration::MigrationApplyApproval>,
marker: PhantomData<fn() -> S>,
}
impl<S: Schema> MigrationApprovalSet<S> {
#[must_use]
pub fn len(&self) -> usize {
self.approvals.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.approvals.is_empty()
}
}
#[derive(Clone, Debug)]
pub struct MigrationPlan<S: Schema> {
#[cfg_attr(not(feature = "typedb"), allow(dead_code))]
catalog: type_bridge_schema_migration::MigrationCatalog,
plan: MigrationPreviewInner,
marker: PhantomData<fn() -> S>,
}
impl<S: Schema> MigrationPlan<S> {
#[must_use]
pub fn execution_authorized(&self) -> bool {
match &self.plan {
MigrationPreviewInner::Apply(plan) => plan.execution_authorized(),
MigrationPreviewInner::Rollback(plan) => plan.execution_authorized(),
}
}
#[cfg(feature = "typedb")]
pub async fn execute(
&self,
database: &Database<S>,
holder: &str,
) -> Result<type_bridge_schema_migration::MigrationExecutionReport> {
self.execute_controlled(
database,
holder,
&type_bridge_schema_migration::MigrationExecutionControl::default(),
)
.await
}
#[cfg(feature = "typedb")]
pub async fn execute_controlled(
&self,
database: &Database<S>,
holder: &str,
control: &type_bridge_schema_migration::MigrationExecutionControl,
) -> Result<type_bridge_schema_migration::MigrationExecutionReport> {
let holder = type_bridge_schema_migration::LeaseHolderId::new(holder)
.map_err(migration_diagnostic)?;
let database = Arc::new(database.inner_orm().clone());
match &self.plan {
MigrationPreviewInner::Apply(plan) => {
type_bridge_schema_migration_typedb::execute_catalog_apply_plan_controlled(
database,
&self.catalog,
&holder,
plan,
control,
)
.await
.map(type_bridge_schema_migration::MigrationExecutionReport::from_apply)
.map_err(migration_diagnostic)
}
MigrationPreviewInner::Rollback(plan) => {
type_bridge_schema_migration_typedb::execute_catalog_rollback_plan_controlled(
database,
&self.catalog,
&holder,
plan,
control,
)
.await
.map(type_bridge_schema_migration::MigrationExecutionReport::from_rollback)
.map_err(migration_diagnostic)
}
}
}
}
#[derive(Clone, Copy, Debug)]
pub struct MigrationPreviewEntry<'a> {
id: &'a MigrationId,
safety: SafetyClass,
step_count: usize,
transaction_group_count: usize,
backfill_count: usize,
reversible: bool,
}
impl MigrationPreviewEntry<'_> {
pub const fn id(&self) -> &MigrationId {
self.id
}
pub const fn safety(&self) -> SafetyClass {
self.safety
}
pub const fn step_count(&self) -> usize {
self.step_count
}
pub const fn transaction_group_count(&self) -> usize {
self.transaction_group_count
}
pub const fn backfill_count(&self) -> usize {
self.backfill_count
}
pub const fn is_reversible(&self) -> bool {
self.reversible
}
}
fn preview_entry(plan: &MigrationPreviewInner, index: usize) -> Option<MigrationPreviewEntry<'_>> {
match plan {
MigrationPreviewInner::Apply(plan) => {
plan.migrations()
.get(index)
.map(|entry| MigrationPreviewEntry {
id: entry.manifest().id(),
safety: entry.manifest().safety(),
step_count: entry.steps().len(),
transaction_group_count: entry.transaction_groups().len(),
backfill_count: entry.backfill_step_indices().len(),
reversible: entry.manifest().reversible(),
})
}
MigrationPreviewInner::Rollback(plan) => {
plan.rollbacks()
.get(index)
.map(|entry| MigrationPreviewEntry {
id: entry.manifest().id(),
safety: entry.rollback_safety(),
step_count: entry.operations().len(),
transaction_group_count: entry.steps().len(),
backfill_count: entry.backfills().len(),
reversible: true,
})
}
}
}
#[derive(Clone, Copy, Debug)]
pub struct MigrationHistoryEntry<'a> {
inner: &'a type_bridge_schema_migration::VerifiedMigrationHistoryBundleEntry,
}
impl MigrationHistoryEntry<'_> {
#[must_use]
pub const fn id(&self) -> &MigrationId {
self.inner.manifest().id()
}
#[must_use]
pub fn parents(&self) -> &[MigrationId] {
self.inner.manifest().parents()
}
#[must_use]
pub const fn manifest_digest(
&self,
) -> type_bridge_contract::migration::MigrationManifestDigest {
self.inner.manifest_digest()
}
#[must_use]
pub fn step_count(&self) -> usize {
self.inner.manifest().steps().len()
}
#[must_use]
pub fn step(&self, index: usize) -> Option<&MigrationStep> {
self.inner.manifest().steps().get(index)
}
#[must_use]
pub const fn safety(&self) -> SafetyClass {
self.inner.manifest().safety()
}
#[must_use]
pub const fn is_reversible(&self) -> bool {
self.inner.manifest().reversible()
}
}
impl<S: Schema> SchemaPackage<S> {
pub fn open_migration_catalog(&self, bytes: &[u8]) -> Result<MigrationCatalog<S>> {
let (_projection, authority) = self.verify_and_install_with_authority()?;
let authority = authority.ok_or_else(|| Error::SchemaVerification {
message: "migration catalogs require generated schema authority".to_owned(),
source: None,
})?;
let context = ManagedDeltaContext::new(
authority.managed_scope().id().clone(),
authority.semantic_profile().id().clone(),
type_bridge_schema_migration::migration_runtime_capability_vocabulary().map_err(
|error| Error::SchemaVerification {
message: format!(
"migration runtime capability vocabulary is invalid [{}]",
error.code().as_str()
),
source: Some(Box::new(error)),
},
)?,
);
let inner = type_bridge_schema_migration::MigrationCatalog::open(bytes, &context).map_err(
|error| Error::SchemaVerification {
message: format!(
"generated migration catalog was rejected [{}]",
error.code().as_str()
),
source: Some(Box::new(error)),
},
)?;
Ok(MigrationCatalog {
inner,
marker: PhantomData,
})
}
}
fn plan_error(error: type_bridge_schema_migration::MigrationApplyPlanError) -> Error {
let error = match error {
type_bridge_schema_migration::MigrationApplyPlanError::Contract(diagnostic) => {
return Error::from_contract_diagnostic(diagnostic);
}
error => error,
};
let message = match &error {
type_bridge_schema_migration::MigrationApplyPlanError::Contract(_) => unreachable!(),
type_bridge_schema_migration::MigrationApplyPlanError::Schema(_) => {
"generated migration preview schema replay failed".to_owned()
}
type_bridge_schema_migration::MigrationApplyPlanError::Lowering(diagnostic) => format!(
"generated migration preview lowering was rejected [{}]",
diagnostic.code()
),
};
Error::SchemaVerification {
message,
source: Some(Box::new(error)),
}
}
fn migration_error(code: &str) -> Error {
Error::SchemaVerification {
message: format!("generated migration operation was rejected [{code}]"),
source: None,
}
}
fn migration_diagnostic(error: type_bridge_contract::diagnostic::Diagnostic) -> Error {
Error::from_contract_diagnostic(error)
}