use std::collections::BTreeSet;
use std::ffi::OsString;
use std::fs;
use std::path::{Component, Path, PathBuf};
use clap::{Parser, Subcommand};
#[cfg(test)]
use type_bridge_schema::SystemSchemaSourceService;
use type_bridge_schema_migration::MigrationGenerationOutcome;
use type_bridge_schema_migration_typedb::execution_capability_vocabulary;
use type_bridge_workspace::{
ConfigOrigin, ExtensionRegistryService, ExtensionRequirement, SecretReference,
SecretReferenceService, TypeBridgeConfigSpec, TypeBridgeWorkspace, TypeBridgeWorkspaceServices,
WorkspaceDirectoryAuthority, WorkspaceEnvironment, WorkspaceRoot, WorkspaceServiceError,
WorkspaceTransportPolicy,
};
#[derive(Parser)]
#[command(
name = "type-bridge",
version,
about = "TypeBridge V2 workspace commands"
)]
struct Cli {
#[arg(long, global = true, default_value = "typebridge.yaml")]
manifest: PathBuf,
#[command(subcommand)]
command: Command,
}
#[derive(Subcommand)]
enum Command {
Schema {
#[command(subcommand)]
command: SchemaCommand,
},
Migration {
#[command(subcommand)]
command: MigrationCommand,
},
}
#[derive(Subcommand)]
enum SchemaCommand {
Check,
Generate,
ExportDeclared {
#[arg(long, default_value = "declared-schema.json")]
output: PathBuf,
},
}
#[derive(Subcommand)]
enum MigrationCommand {
Make {
#[arg(long)]
name: String,
},
Plan,
Apply {
#[arg(long)]
environment: String,
#[arg(long = "approve")]
approvals: Vec<String>,
},
Verify {
#[arg(long)]
environment: String,
},
Adopt {
#[arg(long)]
environment: String,
#[arg(long)]
legacy_directory: PathBuf,
#[arg(long, default_value = "0000_legacy_frontier")]
name: String,
},
}
struct DeferSecrets;
impl SecretReferenceService for DeferSecrets {
fn validate_reference(
&self,
_reference: &SecretReference,
) -> Result<(), WorkspaceServiceError> {
Ok(())
}
}
struct NoExtensions;
impl ExtensionRegistryService for NoExtensions {
fn validate_requirement(
&self,
_requirement: &ExtensionRequirement,
) -> Result<(), WorkspaceServiceError> {
Err(WorkspaceServiceError::new(
"extension_handlers_unavailable_in_cli",
))
}
}
pub fn run_cli<I, T>(arguments: I) -> i32
where
I: IntoIterator<Item = T>,
T: Into<OsString> + Clone,
{
let cli = match Cli::try_parse_from(arguments) {
Ok(cli) => cli,
Err(error) => {
let _ = error.print();
return if error.use_stderr() { 2 } else { 0 };
}
};
match run(&cli) {
Ok(()) => 0,
Err(message) => {
eprintln!("error: {message}");
1
}
}
}
fn run(cli: &Cli) -> Result<(), String> {
let workspace = load_workspace(&cli.manifest)?;
match &cli.command {
Command::Schema {
command: SchemaCommand::Check,
} => {
println!(
"schema sources are valid\n declared identity: {}\n managed semantics: {}",
workspace
.declared_schema()
.declared_identity_fingerprint()
.as_fingerprint()
.digest()
.to_hex(),
workspace
.managed_state()
.managed_semantic_schema()
.as_fingerprint()
.digest()
.to_hex(),
);
Ok(())
}
Command::Schema {
command: SchemaCommand::Generate,
} => run_schema_generate(&workspace),
Command::Schema {
command: SchemaCommand::ExportDeclared { output },
} => run_schema_export_declared(&workspace, output),
Command::Migration { command } => match command {
MigrationCommand::Make { name } => {
let directory = workspace.open_migration_directory().map_err(display)?;
match workspace
.migration_make_in(&directory, name)
.map_err(display)?
{
MigrationGenerationOutcome::UpToDate => {
println!("history already reaches the desired schema");
}
MigrationGenerationOutcome::Generated(generated) => {
workspace
.write_generated_migration_in(&directory, &generated)
.map_err(display)?;
let path = directory.display_path().join(generated.file_name());
println!(
"wrote {}\n safety: {:?}\n preview: {}",
path.display(),
generated.manifest().safety(),
path.with_file_name(generated.preview_file_name()).display(),
);
}
}
Ok(())
}
MigrationCommand::Apply {
environment,
approvals,
} => run_connected(
&workspace,
environment,
ConnectedAction::Apply {
approvals: approvals.clone(),
},
),
MigrationCommand::Verify { environment } => {
run_connected(&workspace, environment, ConnectedAction::Verify)
}
MigrationCommand::Adopt {
environment,
legacy_directory,
name,
} => run_connected(
&workspace,
environment,
ConnectedAction::Adopt {
legacy_directory: legacy_directory.clone(),
name: name.clone(),
},
),
MigrationCommand::Plan => {
let directory = workspace.open_migration_directory().map_err(display)?;
let plan = workspace
.migration_plan_in(&directory, &BTreeSet::new())
.map_err(display)?;
if plan.is_empty() {
println!("no committed migrations");
return Ok(());
}
for entry in plan {
println!(
"{}/{} safety={:?} reversible={}",
entry.id().app_label().as_str(),
entry.id().name().as_str(),
entry.safety(),
entry.reversible(),
);
}
Ok(())
}
},
}
}
fn load_workspace(manifest: &PathBuf) -> Result<TypeBridgeWorkspace, String> {
let manifest = fs::canonicalize(manifest)
.map_err(|error| format!("cannot resolve {}: {error}", manifest.display()))?;
let root = manifest
.parent()
.ok_or_else(|| "workspace manifest has no parent directory".to_owned())?;
let file_name = manifest
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| "workspace manifest has no UTF-8 file name".to_owned())?;
let root = WorkspaceRoot::new(root).map_err(display)?;
let source = WorkspaceDirectoryAuthority::open(root.clone()).map_err(display)?;
let origin = ConfigOrigin::new(root, file_name, "type-bridge cli").map_err(display)?;
let limit = type_bridge_contract::limits::MAX_CANONICAL_BYTES;
let captured = source
.capture_relative_file(Path::new(file_name), limit)
.map_err(|error| format!("cannot read {}: {error}", manifest.display()))?;
let bytes = captured.bytes();
if bytes.len() > limit {
return Err(format!(
"{} exceeds the 16 MiB manifest ceiling",
manifest.display()
));
}
let located = TypeBridgeConfigSpec::from_yaml_bytes(bytes, origin).map_err(display)?;
let available = execution_capability_vocabulary().map_err(display)?;
let secrets = DeferSecrets;
let extensions = NoExtensions;
let services = TypeBridgeWorkspaceServices::new(&source, &secrets, &extensions, &available);
TypeBridgeWorkspace::from_located_config(located, &services).map_err(display)
}
fn display(error: impl std::fmt::Display) -> String {
error.to_string()
}
fn sanitize_connected_error(
context: String,
code: &'static str,
error: type_bridge_orm::SecureConnectError,
) -> String {
match error.credential_safe_diagnostic() {
Some(diagnostic) => format!("{context}: {diagnostic}"),
None => format!("{context} [{code}]; inspect provider logs"),
}
}
fn sanitize_schema_export_error(_error: type_bridge_orm::OrmError) -> String {
"cannot export the managed schema [typedb_schema_export_failed]; inspect provider logs"
.to_owned()
}
fn sanitize_migration_execution_outcome(
context: &str,
outcome: type_bridge_schema_migration::MigrationExecutionOutcome,
) -> String {
use type_bridge_schema_migration::{
MigrationExecutionOutcome as Outcome, MigrationExecutionPosition as Position,
};
let render_position = |position| match position {
Position::TransactionGroup(ordinal) => format!("transaction group {ordinal}"),
Position::ManifestCheckpoint => "manifest checkpoint".to_owned(),
};
match outcome {
Outcome::Applied => format!("{context}: applied"),
Outcome::RetrySafe {
migration_id,
position,
diagnostic,
} => format!(
"{context}: retry-safe at {}/{} ({position}): {diagnostic}",
migration_id.app_label().as_str(),
migration_id.name().as_str(),
position = render_position(position),
),
Outcome::RequiresExplicitRecovery {
migration_id,
position,
diagnostic,
} => format!(
"{context}: explicit recovery required at {}/{} ({position}): {diagnostic}",
migration_id.app_label().as_str(),
migration_id.name().as_str(),
position = render_position(position),
),
}
}
fn run_schema_generate(workspace: &TypeBridgeWorkspace) -> Result<(), String> {
use type_bridge_contract::projection::{BindingTarget, ProjectionConfig};
use type_bridge_schema::project;
use type_bridge_schema_codegen::{PythonEmitter, RustEmitter, TypeScriptEmitter};
let outputs = workspace.config().outputs();
if outputs.is_empty() {
return Err(
"no binding outputs configured; add bindings.<target>.output to the manifest".into(),
);
}
let resolved = workspace.resolved_schema();
for (target, directory) in outputs {
let package = match target {
BindingTarget::Python => {
let emitter = PythonEmitter::new();
let projection = project(
resolved,
BindingTarget::Python,
&ProjectionConfig::python(),
&emitter.generator_handlers(),
&emitter.code_resources().map_err(display)?,
)
.map_err(display)?;
emitter.emit(&projection)
}
BindingTarget::TypeScript => {
let emitter = TypeScriptEmitter::new();
let projection = project(
resolved,
BindingTarget::TypeScript,
&ProjectionConfig::typescript(),
&emitter.generator_handlers(),
&emitter.code_resources().map_err(display)?,
)
.map_err(display)?;
emitter.emit(&projection)
}
BindingTarget::Rust => {
let emitter = RustEmitter::new();
let projection = project(
resolved,
BindingTarget::Rust,
&ProjectionConfig::rust(),
&emitter.generator_handlers(),
&emitter.code_resources().map_err(display)?,
)
.map_err(display)?;
emitter.emit_with_declared_schema(&projection, workspace.declared_schema())
}
}
.map_err(display)?;
let workspace_root = workspace.output_root()?;
let root = workspace_root.open_beneath(directory.as_path())?;
for (path, bytes) in package.files() {
let relative = std::path::Path::new(path);
validate_generated_relative_path(relative)?;
let parent = root.open_beneath(
relative
.parent()
.unwrap_or_else(|| std::path::Path::new("")),
)?;
let file_name = relative
.file_name()
.ok_or_else(|| format!("generated path {path:?} has no file name"))?;
parent.write_atomic(file_name, bytes)?;
}
println!(
"generated {} file(s) for {} into {}",
package.files().len(),
match target {
BindingTarget::Python => "python",
BindingTarget::TypeScript => "typescript",
BindingTarget::Rust => "rust",
},
root.display_path().display(),
);
}
Ok(())
}
fn run_schema_export_declared(
workspace: &TypeBridgeWorkspace,
output: &Path,
) -> Result<(), String> {
use type_bridge_contract::schema::encode_declared_schema;
validate_declared_output_path(output)?;
let root = workspace.output_root()?;
let parent = root.open_beneath(output.parent().unwrap_or_else(|| std::path::Path::new("")))?;
let file_name = output
.file_name()
.ok_or_else(|| "declared-schema output has no file name".to_owned())?;
let destination = parent.display_path().join(file_name);
let bytes = encode_declared_schema(workspace.declared_schema()).map_err(display)?;
parent.write_atomic(file_name, &bytes)?;
println!(
"wrote canonical declared schema to {}\n declared identity: {}",
destination.display(),
workspace
.declared_schema()
.declared_identity_fingerprint()
.as_fingerprint()
.digest()
.to_hex(),
);
Ok(())
}
fn validate_declared_output_path(output: &Path) -> Result<(), String> {
let Some(portable) = output.to_str() else {
return Err("declared-schema output must be valid UTF-8".into());
};
let invalid_spelling = portable.is_empty()
|| portable.contains(['\\', ':', '\0'])
|| portable.bytes().any(|byte| byte.is_ascii_control())
|| portable
.split('/')
.any(|segment| segment.is_empty() || matches!(segment, "." | ".."));
let invalid_components = output.is_absolute()
|| output
.components()
.any(|component| !matches!(component, Component::Normal(_)));
if invalid_spelling || invalid_components {
return Err("declared-schema output must be a confined portable workspace path".into());
}
if output.extension().and_then(|extension| extension.to_str()) != Some("json") {
return Err("declared-schema output must end in lowercase .json".into());
}
Ok(())
}
fn validate_generated_relative_path(path: &std::path::Path) -> Result<(), String> {
if path.as_os_str().is_empty()
|| path
.components()
.any(|component| !matches!(component, std::path::Component::Normal(_)))
{
return Err(format!(
"generated output path {:?} is not a confined relative file",
path
));
}
Ok(())
}
enum ConnectedAction {
Apply {
approvals: Vec<String>,
},
Verify,
Adopt {
legacy_directory: PathBuf,
name: String,
},
}
fn secure_connect_options(
environment: &WorkspaceEnvironment,
) -> type_bridge_orm::SecureConnectOptions {
let tls_mode = match environment.transport_policy() {
WorkspaceTransportPolicy::Disabled => type_bridge_orm::TlsMode::Disabled,
WorkspaceTransportPolicy::NativeRoots => type_bridge_orm::TlsMode::NativeRoots,
WorkspaceTransportPolicy::CustomRootCa(root_ca) => {
type_bridge_orm::TlsMode::CustomRootCa(root_ca.as_path().to_path_buf())
}
};
let mut options = type_bridge_orm::SecureConnectOptions {
tls_mode,
..type_bridge_orm::SecureConnectOptions::default()
};
if let Some(port) = environment.http_port() {
options.http_port = port;
}
options
}
fn preflight_secure_connect_options(
workspace: &TypeBridgeWorkspace,
environment_name: &str,
) -> Result<type_bridge_orm::PreparedSecureConnectOptions, String> {
let environment = workspace
.config()
.environment(environment_name)
.ok_or_else(|| {
format!("environment {environment_name:?} is not owned by this workspace")
})?;
let options = secure_connect_options(environment);
match workspace
.capture_environment_custom_root_ca(environment_name)
.map_err(display)?
{
Some(bytes) => options
.prepare_transport_from_captured_custom_root(bytes)
.map_err(display),
None => options.prepare_transport().map_err(display),
}
}
fn run_connected(
workspace: &TypeBridgeWorkspace,
environment: &str,
action: ConnectedAction,
) -> Result<(), String> {
let runtime = tokio::runtime::Runtime::new()
.map_err(|error| format!("cannot start the async runtime: {error}"))?;
runtime.block_on(run_connected_async(workspace, environment, action))
}
async fn run_connected_async(
workspace: &TypeBridgeWorkspace,
environment_name: &str,
action: ConnectedAction,
) -> Result<(), String> {
let config = workspace.config();
let Some(environment) = config.environment(environment_name) else {
let known = config
.environments()
.keys()
.cloned()
.collect::<Vec<_>>()
.join(", ");
return Err(format!(
"unknown environment {environment_name:?}; the manifest declares: [{known}]"
));
};
if matches!(
&action,
ConnectedAction::Apply { .. } | ConnectedAction::Adopt { .. }
) && !environment.migrate()
{
return Err(format!(
"environment {environment_name:?} is not opted into migration \
application; set `migrate: true` in the manifest to allow it"
));
}
environment
.requirements()
.ensure_supported_by(&execution_capability_vocabulary().map_err(display)?)
.map_err(display)?;
let prepared_adoption = match &action {
ConnectedAction::Adopt {
legacy_directory,
name,
} => Some(prepare_legacy_adoption(workspace, legacy_directory, name)?),
ConnectedAction::Apply { .. } | ConnectedAction::Verify => None,
};
let migration_directory = if matches!(&action, ConnectedAction::Adopt { .. }) {
workspace.ensure_migration_directory().map_err(display)?
} else {
workspace.open_migration_directory().map_err(display)?
};
let ordinary_graph = if prepared_adoption.is_none() {
Some(
workspace
.discover_migrations_in(&migration_directory)
.map_err(display)?,
)
} else {
None
};
let prepared_approvals = match &action {
ConnectedAction::Apply { approvals } => Some(bind_approvals(
ordinary_graph
.as_ref()
.ok_or_else(|| "internal apply history was not retained".to_owned())?,
approvals,
)?),
ConnectedAction::Verify | ConnectedAction::Adopt { .. } => None,
};
let options = preflight_secure_connect_options(workspace, environment_name)?;
let username = resolve_credential(environment.username())?;
let password = resolve_credential(environment.password())?;
let journal_name =
type_bridge_schema_migration_typedb::derived_journal_database_name(environment.database());
let managed_requires_existing = match &action {
ConnectedAction::Verify => Some(
"`migration verify` is read-only and never creates databases \
— apply migrations to this environment first",
),
ConnectedAction::Adopt { .. } => {
Some("`migration adopt` cutover requires the migrated v1 database to already exist")
}
ConnectedAction::Apply { .. } => None,
};
if let Some(reason) = managed_requires_existing {
let exists = type_bridge_orm::database_exists_prepared_secure(
environment.uri(),
environment.database(),
&username,
&password,
options.clone(),
)
.await
.map_err(|error| {
sanitize_connected_error(
format!("cannot check database {:?}", environment.database()),
"typedb_database_exists_failed",
error,
)
})?;
if !exists {
return Err(format!(
"database {:?} does not exist; {reason}",
environment.database()
));
}
} else {
type_bridge_orm::ensure_database_exists_prepared_secure(
environment.uri(),
environment.database(),
&username,
&password,
options.clone(),
)
.await
.map_err(|error| {
sanitize_connected_error(
format!("cannot ensure database {:?}", environment.database()),
"typedb_database_ensure_failed",
error,
)
})?;
}
let managed = std::sync::Arc::new(
type_bridge_orm::Database::connect_prepared_secure_with_options(
environment.uri(),
environment.database(),
&username,
&password,
options.clone(),
)
.await
.map_err(|error| {
sanitize_connected_error(
"cannot connect the managed database".to_owned(),
"typedb_database_connect_failed",
error,
)
})?,
);
let adoption_files = if let Some(prepared) = prepared_adoption.as_ref() {
verify_prepared_adoption_live(&managed, prepared).await?;
Some(publish_prepared_adoption(
workspace,
&migration_directory,
prepared,
)?)
} else {
None
};
if matches!(&action, ConnectedAction::Verify) {
let exists = type_bridge_orm::database_exists_prepared_secure(
environment.uri(),
&journal_name,
&username,
&password,
options.clone(),
)
.await
.map_err(|error| {
sanitize_connected_error(
format!("cannot check database {journal_name:?}"),
"typedb_database_exists_failed",
error,
)
})?;
if !exists {
return Err(format!(
"database {journal_name:?} does not exist; `migration verify` is read-only and never creates databases"
));
}
} else {
type_bridge_orm::ensure_database_exists_prepared_secure(
environment.uri(),
&journal_name,
&username,
&password,
options.clone(),
)
.await
.map_err(|error| {
sanitize_connected_error(
format!("cannot ensure database {journal_name:?}"),
"typedb_database_ensure_failed",
error,
)
})?;
}
let journal = std::sync::Arc::new(
type_bridge_orm::Database::connect_prepared_secure_with_options(
environment.uri(),
&journal_name,
&username,
&password,
options,
)
.await
.map_err(|error| {
sanitize_connected_error(
"cannot connect the journal database".to_owned(),
"typedb_database_connect_failed",
error,
)
})?,
);
let genesis = workspace
.migration_genesis_in(&migration_directory)
.map_err(display)?;
let lowering = type_bridge_schema_migration::SchemaLoweringBinding::current(
workspace.delta_context().available_capabilities().clone(),
)
.map_err(display)?;
let runner = type_bridge_schema_migration_typedb::TypeDbMigrationRunner::new(
managed,
journal,
genesis.clone(),
workspace.delta_context().clone(),
lowering,
config.migration_policy().clone(),
);
let holder =
type_bridge_schema_migration::LeaseHolderId::new("type-bridge-cli").map_err(display)?;
let directory = migration_directory.directory();
match action {
ConnectedAction::Apply { .. } => {
let approvals = prepared_approvals
.as_deref()
.ok_or_else(|| "internal apply approvals were not retained".to_owned())?;
let outcome = runner
.apply_in(
directory,
&type_bridge_schema_migration::MigrationApplyTarget::DefaultHead,
&holder,
approvals,
)
.await
.map_err(display)?;
match outcome {
type_bridge_schema_migration_typedb::MigrationDirectoryApplyOutcome::UpToDate => {
println!("applied ledger already reaches the committed head");
Ok(())
}
type_bridge_schema_migration_typedb::MigrationDirectoryApplyOutcome::Executed(
type_bridge_schema_migration::MigrationExecutionOutcome::Applied,
) => {
println!("applied the committed chain");
Ok(())
}
type_bridge_schema_migration_typedb::MigrationDirectoryApplyOutcome::Executed(
outcome,
) => Err(sanitize_migration_execution_outcome(
"apply did not complete",
outcome,
)),
}
}
ConnectedAction::Verify => {
let report = runner
.verify_in(directory, Some(workspace.declared_schema()))
.await
.map_err(display)?;
if report.is_clean() {
println!(
"migration state is coherent\n applied frontier: {}",
report
.applied_frontier()
.iter()
.map(|id| format!("{}/{}", id.app_label().as_str(), id.name().as_str()))
.collect::<Vec<_>>()
.join(", "),
);
Ok(())
} else {
for finding in report.findings() {
eprintln!("drift: {finding:?}");
}
Err(format!("{} drift finding(s)", report.findings().len()))
}
}
ConnectedAction::Adopt { .. } => {
let bridge_display_path = adoption_files
.ok_or_else(|| "internal adoption preflight state was not retained".to_owned())?;
let prepared = prepared_adoption
.as_ref()
.ok_or_else(|| "internal adoption authority was not retained".to_owned())?;
let outcome = runner
.import_verified_legacy_frontier_in(
&prepared.history,
&prepared.reconstructed,
directory,
&holder,
)
.await;
match outcome {
Ok(
type_bridge_schema_migration_typedb::MigrationDirectoryApplyOutcome::UpToDate,
) => {
println!("legacy history is already adopted; the bridged ledger is current");
Ok(())
}
Ok(
type_bridge_schema_migration_typedb::MigrationDirectoryApplyOutcome::Executed(
type_bridge_schema_migration::MigrationExecutionOutcome::Applied,
),
) => {
println!(
"adopted the legacy history\n genesis: {}\n bridge: {}",
migration_directory
.display_path()
.join(type_bridge_schema_compat::ADOPTED_GENESIS_FILE_NAME)
.display(),
bridge_display_path.display(),
);
Ok(())
}
Ok(
type_bridge_schema_migration_typedb::MigrationDirectoryApplyOutcome::Executed(
outcome,
),
) => Err(sanitize_migration_execution_outcome(
"adoption checkpoint did not complete",
outcome,
)),
Err(error) => Err(display(error)),
}
}
}
}
struct PreparedLegacyAdoption {
history: type_bridge_migration::LegacyAdoptionHistory,
reconstructed: type_bridge_migration::VerifiedLegacyHead,
authority: type_bridge_schema_compat::AdoptedGenesisAuthority,
bridge: type_bridge_schema_migration::VerifiedSchemaMigrationManifest,
bridge_name: String,
bridge_bytes: Vec<u8>,
}
fn prepare_legacy_adoption(
workspace: &TypeBridgeWorkspace,
legacy_directory: &std::path::Path,
name: &str,
) -> Result<PreparedLegacyAdoption, String> {
let migration_name =
type_bridge_contract::migration::MigrationName::new(name.to_owned()).map_err(display)?;
let bridge_name = format!("{}.tbmigration.json", migration_name.as_str());
let history =
type_bridge_migration::load_adoption_history(legacy_directory).map_err(|error| {
format!("legacy migration directory failed the checked adoption loader: {error}")
})?;
let reconstructed = type_bridge_migration::reconstruct_legacy_head(&history)
.map_err(|error| format!("legacy head reconstruction failed: {error}"))?;
let authority = type_bridge_schema_compat::parse_adopted_genesis_authority(
type_bridge_contract::schema::DocumentId::new("legacy-head-snapshot.typeql")
.map_err(display)?,
reconstructed.schema_typeql(),
)
.map_err(display)?;
let frontier = type_bridge_schema_migration_typedb::extract_legacy_frontier(history.graph())
.map_err(display)?;
let applied_set =
type_bridge_schema_migration_typedb::extract_legacy_applied_set_digest(history.graph())
.map_err(display)?;
let id = type_bridge_contract::migration::MigrationId::from_components(
type_bridge_contract::migration::MigrationAppLabel::new(
workspace.config().app_label().as_str().to_owned(),
)
.map_err(display)?,
migration_name,
);
let bridge = type_bridge_schema_migration::build_legacy_frontier_bridge(
id,
frontier,
applied_set,
authority.declared(),
workspace.delta_context(),
)
.map_err(display)?;
let bridge_bytes =
type_bridge_schema_migration::encode_verified_manifest(&bridge).map_err(display)?;
history
.require_unchanged_head(&reconstructed)
.map_err(|error| {
format!("legacy migration directory changed during adoption preparation: {error}")
})?;
Ok(PreparedLegacyAdoption {
history,
reconstructed,
authority,
bridge,
bridge_name,
bridge_bytes,
})
}
async fn verify_prepared_adoption_live(
managed: &type_bridge_orm::Database,
prepared: &PreparedLegacyAdoption,
) -> Result<(), String> {
let export = managed
.schema_text()
.await
.map_err(sanitize_schema_export_error)?;
prepared
.history
.require_unchanged_head(&prepared.reconstructed)
.map_err(|error| format!("legacy adoption history changed during live export: {error}"))?;
let expected_internal = type_bridge_schema_compat::released_typeql_to_declared_projection(
type_bridge_contract::schema::DocumentId::new("managed-fence-schema.typeql")
.map_err(display)?,
type_bridge_schema_migration_typedb::MANAGED_FENCE_SCHEMA_TYPEQL,
)
.map_err(display)?;
let live = type_bridge_schema_compat::parse_adopted_genesis_authority_with_internal(
type_bridge_contract::schema::DocumentId::new("legacy-live-head.typeql")
.map_err(display)?,
&export,
Some(&expected_internal),
)
.map_err(display)?;
if live.legacy_identity() != prepared.authority.legacy_identity()
|| live.declared().declared_identity_fingerprint()
!= prepared
.authority
.declared()
.declared_identity_fingerprint()
|| live.released_extension_identity() != prepared.authority.released_extension_identity()
{
return Err(
"live managed schema differs from the independently verified legacy-head snapshot"
.to_owned(),
);
}
Ok(())
}
fn publish_prepared_adoption(
workspace: &TypeBridgeWorkspace,
migration_directory: &type_bridge_workspace::MigrationDirectoryAuthority,
prepared: &PreparedLegacyAdoption,
) -> Result<PathBuf, String> {
publish_prepared_adoption_with_after_bridge(workspace, migration_directory, prepared, || {})
}
fn publish_prepared_adoption_with_after_bridge<F>(
workspace: &TypeBridgeWorkspace,
migration_directory: &type_bridge_workspace::MigrationDirectoryAuthority,
prepared: &PreparedLegacyAdoption,
after_bridge: F,
) -> Result<PathBuf, String>
where
F: FnOnce(),
{
let directory = migration_directory.directory();
let _lock = directory.try_acquire_authoring_lock().map_err(|error| {
if error.kind() == std::io::ErrorKind::WouldBlock {
"migration adoption conflicts with another canonical history publisher".to_owned()
} else {
format!("cannot lock canonical migration publication: {error}")
}
})?;
let genesis_name = type_bridge_schema_compat::ADOPTED_GENESIS_FILE_NAME;
let genesis_bytes = prepared.reconstructed.schema_typeql().as_bytes();
let mut bridge_created = false;
let mut genesis_created = false;
let mut after_bridge = Some(after_bridge);
let publication = (|| -> Result<(), String> {
if let Some(existing) = read_existing_authority(directory, genesis_name)?
&& existing != genesis_bytes
{
return Err(format!(
"{genesis_name} already exists but differs from the verified legacy-head snapshot"
));
}
let bridge_already_published =
if let Some(existing) = read_existing_authority(directory, &prepared.bridge_name)? {
if existing != prepared.bridge_bytes {
return Err(format!(
"{} already exists with different authority bytes",
prepared.bridge_name
));
}
true
} else {
false
};
let (current, evidence) =
type_bridge_schema_migration::discover_verified_migration_chain_with_evidence_in(
directory,
prepared.authority.declared(),
workspace.delta_context(),
)
.map_err(display)?;
let prospective = if current.manifest(prepared.bridge.id()).is_some() {
current
} else {
let manifests = current
.manifests()
.map(|(_, manifest)| manifest.clone())
.chain(std::iter::once(prepared.bridge.clone()))
.collect::<Vec<_>>();
type_bridge_schema_migration::MigrationHistoryGraph::from_verified(manifests)
.map_err(display)?
};
type_bridge_schema_migration::require_adoption_authority_pair(&prospective, true)
.map_err(display)?;
evidence.require_unchanged(directory).map_err(display)?;
prepared
.history
.require_unchanged_head(&prepared.reconstructed)
.map_err(|error| {
format!("legacy adoption history changed before pair publication: {error}")
})?;
if !bridge_already_published {
bridge_created =
publish_authority(directory, &prepared.bridge_name, &prepared.bridge_bytes)?;
}
if let Some(after_bridge) = after_bridge.take() {
after_bridge();
}
prepared
.history
.require_unchanged_head(&prepared.reconstructed)
.map_err(|error| {
format!("legacy adoption history changed before genesis publication: {error}")
})?;
genesis_created = publish_authority(directory, genesis_name, genesis_bytes)?;
prepared
.history
.require_unchanged_head(&prepared.reconstructed)
.map_err(|error| {
format!("legacy adoption history changed after pair publication: {error}")
})?;
workspace
.discover_migrations_in(migration_directory)
.map_err(display)?;
Ok(())
})();
if let Err(error) = publication {
return Err(rollback_adoption_publication(
directory,
&prepared.bridge_name,
bridge_created,
genesis_created,
error,
));
}
Ok(migration_directory
.display_path()
.join(&prepared.bridge_name))
}
fn rollback_adoption_publication(
directory: &type_bridge_schema_migration::MigrationDirectory,
bridge_name: &str,
bridge_created: bool,
genesis_created: bool,
primary: String,
) -> String {
let mut cleanup_errors = Vec::new();
if genesis_created
&& let Err(error) =
directory.remove_file(type_bridge_schema_compat::ADOPTED_GENESIS_FILE_NAME.as_ref())
{
cleanup_errors.push(format!("cannot remove newly published genesis: {error}"));
}
if bridge_created && let Err(error) = directory.remove_file(bridge_name.as_ref()) {
cleanup_errors.push(format!("cannot remove newly published bridge: {error}"));
}
if (bridge_created || genesis_created)
&& let Err(error) = directory.sync_all()
{
cleanup_errors.push(format!("cannot flush adoption rollback: {error}"));
}
if cleanup_errors.is_empty() {
primary
} else {
format!(
"{primary}; adoption publication rollback failed: {}",
cleanup_errors.join("; ")
)
}
}
fn publish_authority(
directory: &type_bridge_schema_migration::MigrationDirectory,
name: &str,
bytes: &[u8],
) -> Result<bool, String> {
use std::io::Write;
if let Some(existing) = read_existing_authority(directory, name)? {
if existing == bytes {
return Ok(false);
}
return Err(format!(
"{name} already exists with different authority bytes"
));
}
let mut temporary = None;
for attempt in 0..128_u64 {
let candidate = unique_authority_temporary_name(name, attempt);
match directory.create_new(candidate.as_ref()) {
Ok(mut file) => {
if let Err(error) = file.write_all(bytes).and_then(|()| file.sync_all()) {
let _ = directory.remove_file(candidate.as_ref());
return Err(format!("cannot write {candidate}: {error}"));
}
temporary = Some(candidate);
break;
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(error) => {
return Err(format!("cannot create {candidate}: {error}"));
}
}
}
let temporary = temporary.ok_or_else(|| {
format!("cannot allocate a unique temporary authority file beside {name}")
})?;
let publication = directory.hard_link(temporary.as_ref(), name.as_ref());
match publication {
Ok(()) => {
if let Err(error) = directory.sync_all() {
let _ = directory.remove_file(temporary.as_ref());
return Err(format!("cannot flush migration directory: {error}"));
}
let _ = directory.remove_file(temporary.as_ref());
directory
.sync_all()
.map_err(|error| format!("cannot flush migration directory: {error}"))?;
Ok(true)
}
Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
let _ = directory.remove_file(temporary.as_ref());
let existing = read_existing_authority(directory, name)?
.ok_or_else(|| format!("{name} disappeared during no-replace publication"))?;
if existing == bytes {
Ok(false)
} else {
Err(format!(
"{name} was concurrently published with different authority bytes"
))
}
}
Err(error) => {
let _ = directory.remove_file(temporary.as_ref());
Err(format!("cannot publish {name}: {error}"))
}
}
}
fn read_existing_authority(
directory: &type_bridge_schema_migration::MigrationDirectory,
name: &str,
) -> Result<Option<Vec<u8>>, String> {
use std::io::Read;
let file = match directory.open_regular_readonly(name.as_ref()) {
Ok(file) => file,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => return Err(format!("cannot read {name}: {error}")),
};
let limit = type_bridge_contract::limits::MAX_CANONICAL_BYTES;
let mut bytes = Vec::new();
file.take(u64::try_from(limit).unwrap_or(u64::MAX).saturating_add(1))
.read_to_end(&mut bytes)
.map_err(|error| format!("cannot read {name}: {error}"))?;
if bytes.len() > limit {
return Err(format!("{name} exceeds the 16 MiB authority ceiling"));
}
Ok(Some(bytes))
}
fn unique_authority_temporary_name(name: &str, attempt: u64) -> String {
use std::sync::atomic::{AtomicU64, Ordering};
static NEXT_AUTHORITY_TEMPORARY: AtomicU64 = AtomicU64::new(1);
let nonce = NEXT_AUTHORITY_TEMPORARY.fetch_add(1, Ordering::Relaxed);
format!(".{name}.{}.{}.{}.tmp", std::process::id(), nonce, attempt)
}
fn resolve_credential(
reference: &type_bridge_workspace::SecretReference,
) -> Result<String, String> {
std::env::var(reference.environment_variable()).map_err(|_| {
format!(
"credential environment variable {:?} is not set",
reference.environment_variable()
)
})
}
#[cfg(test)]
mod credential_error_redaction_tests {
use super::*;
use type_bridge_contract::diagnostic::{Diagnostic, DiagnosticCategory, DiagnosticCode};
use type_bridge_typedb_runtime::RuntimeError;
const PROVIDER_TEXT: &str =
"TB_ADDRESS_SECRET TB_USERNAME_SECRET TB_PASSWORD_SECRET TB_PROVIDER_SECRET";
const SECRETS: [&str; 4] = [
"TB_ADDRESS_SECRET",
"TB_USERNAME_SECRET",
"TB_PASSWORD_SECRET",
"TB_PROVIDER_SECRET",
];
fn hostile_secure_error() -> type_bridge_orm::SecureConnectError {
type_bridge_orm::SecureConnectError::Runtime(RuntimeError::Connection(
PROVIDER_TEXT.to_owned(),
))
}
#[test]
fn connected_lifecycle_contexts_drop_hostile_provider_text() {
for (context, code) in [
(
"cannot check database \"managed\"",
"typedb_database_exists_failed",
),
(
"cannot ensure database \"managed\"",
"typedb_database_ensure_failed",
),
(
"cannot connect the managed database",
"typedb_database_connect_failed",
),
] {
let sanitized =
sanitize_connected_error(context.to_owned(), code, hostile_secure_error());
let rendered = format!("{sanitized}\n{sanitized:?}");
for secret in SECRETS {
assert!(!rendered.contains(secret), "{secret}: {rendered}");
}
assert!(rendered.contains(context), "{rendered}");
assert!(rendered.contains(code), "{rendered}");
}
}
#[test]
fn connected_lifecycle_preserves_only_typed_safe_diagnostics() {
let sanitized = sanitize_connected_error(
"cannot connect the managed database".to_owned(),
"typedb_database_connect_failed",
type_bridge_orm::SecureConnectError::DriverTlsConfiguration { band: 9 },
);
assert!(
sanitized.contains("tls_driver_lowering_failed"),
"{sanitized}"
);
assert!(sanitized.contains("driver band 9"), "{sanitized}");
assert!(
!sanitized.contains("typedb_database_connect_failed"),
"{sanitized}"
);
}
#[test]
fn schema_export_drops_hostile_orm_text_and_source() {
let sanitized = sanitize_schema_export_error(type_bridge_orm::OrmError::Connection(
PROVIDER_TEXT.into(),
));
let rendered = format!("{sanitized}\n{sanitized:?}");
for secret in SECRETS {
assert!(!rendered.contains(secret), "{secret}: {rendered}");
}
assert!(
rendered.contains("typedb_schema_export_failed"),
"{rendered}"
);
}
#[test]
fn migration_runner_display_omits_provider_details() {
let diagnostic = Diagnostic::new(
DiagnosticCategory::InvalidContract,
DiagnosticCode::new("migration_provider_test_failed").expect("static code"),
"migration provider operation failed",
)
.with_detail("provider", PROVIDER_TEXT);
let error = type_bridge_schema_migration_typedb::MigrationDirectoryApplyError::Diagnostic(
diagnostic,
);
let rendered = display(error);
for secret in SECRETS {
assert!(!rendered.contains(secret), "{secret}: {rendered}");
}
assert!(
rendered.contains("migration_provider_test_failed"),
"{rendered}"
);
}
#[test]
fn migration_outcome_projection_omits_provider_details_for_apply_and_adopt() {
use type_bridge_contract::migration::MigrationId;
use type_bridge_schema_migration::{MigrationExecutionOutcome, MigrationExecutionPosition};
let diagnostic = || {
Diagnostic::new(
DiagnosticCategory::InvalidContract,
DiagnosticCode::new("migration_provider_test_failed").expect("static code"),
"migration provider operation failed",
)
.with_detail("provider", PROVIDER_TEXT)
};
for context in [
"apply did not complete",
"adoption checkpoint did not complete",
] {
for (outcome, expected_state, expected_position) in [
(
MigrationExecutionOutcome::RetrySafe {
migration_id: MigrationId::new("example", "0001_initial")
.expect("migration id"),
position: MigrationExecutionPosition::TransactionGroup(7),
diagnostic: diagnostic(),
},
"retry-safe",
"transaction group 7",
),
(
MigrationExecutionOutcome::RequiresExplicitRecovery {
migration_id: MigrationId::new("example", "0001_initial")
.expect("migration id"),
position: MigrationExecutionPosition::ManifestCheckpoint,
diagnostic: diagnostic(),
},
"explicit recovery required",
"manifest checkpoint",
),
] {
let rendered = sanitize_migration_execution_outcome(context, outcome);
for secret in SECRETS {
assert!(!rendered.contains(secret), "{secret}: {rendered}");
}
for expected in [
context,
expected_state,
"example/0001_initial",
expected_position,
"migration_provider_test_failed",
"migration provider operation failed",
] {
assert!(rendered.contains(expected), "{expected}: {rendered}");
}
}
}
}
}
#[cfg(all(test, unix))]
mod output_authority_tests {
use super::*;
use std::os::unix::fs::symlink;
#[test]
fn retained_output_authority_survives_component_swap_without_redirecting() {
let workspace = tempfile::tempdir().expect("workspace directory");
let outside = tempfile::tempdir().expect("outside directory");
fs::create_dir_all(workspace.path().join("generated/python")).expect("output directory");
let authority = WorkspaceDirectoryAuthority::open(
WorkspaceRoot::new(fs::canonicalize(workspace.path()).expect("canonical workspace"))
.expect("workspace root"),
)
.expect("workspace authority");
let root = authority.output_root().expect("output authority");
let output = root
.open_beneath(Path::new("generated/python"))
.expect("output authority");
let held = workspace.path().join("generated/python-held");
fs::rename(workspace.path().join("generated/python"), &held)
.expect("move retained output directory");
symlink(outside.path(), workspace.path().join("generated/python"))
.expect("redirect configured output path");
output
.write_atomic("_models.py".as_ref(), b"retained authority")
.expect("publication remains handle-relative");
assert_eq!(
fs::read(held.join("_models.py")).expect("retained output reads"),
b"retained authority"
);
assert!(
!outside.path().join("_models.py").exists(),
"component replacement redirected output outside the workspace"
);
}
#[test]
fn retained_output_root_survives_root_entry_swap_without_redirecting() {
let workspace = tempfile::tempdir().expect("workspace directory");
let outside = tempfile::tempdir().expect("outside directory");
fs::create_dir_all(workspace.path().join("generated/python")).expect("output directory");
let authority = WorkspaceDirectoryAuthority::open(
WorkspaceRoot::new(fs::canonicalize(workspace.path()).expect("canonical workspace"))
.expect("workspace root"),
)
.expect("workspace authority");
let root = authority.output_root().expect("output authority");
let held = workspace
.path()
.parent()
.expect("temporary parent")
.join(format!(
"{}-retained",
workspace
.path()
.file_name()
.expect("temporary name")
.to_string_lossy()
));
fs::rename(workspace.path(), &held).expect("workspace root moves after validation");
symlink(outside.path(), workspace.path()).expect("workspace name redirects outside");
let output = root
.open_beneath(Path::new("generated/python"))
.expect("output opens through retained root");
output
.write_atomic("_models.py".as_ref(), b"retained root authority")
.expect("publication remains rooted in the retained handle");
assert_eq!(
fs::read(held.join("generated/python/_models.py")).expect("retained output reads"),
b"retained root authority"
);
assert!(
!outside.path().join("generated/python/_models.py").exists(),
"root replacement redirected output outside the workspace"
);
fs::remove_file(workspace.path()).expect("replacement symlink removes");
fs::rename(&held, workspace.path()).expect("workspace restores for cleanup");
}
}
fn bind_approvals(
graph: &type_bridge_schema_migration::MigrationHistoryGraph,
approvals: &[String],
) -> Result<Vec<type_bridge_schema_migration::MigrationApplyApproval>, String> {
if approvals.is_empty() {
return Ok(Vec::new());
}
approvals
.iter()
.map(|compound| {
let (app_label, name) = compound
.split_once('/')
.ok_or_else(|| format!("approval {compound:?} must be app-label/name"))?;
let id = type_bridge_contract::migration::MigrationId::from_components(
type_bridge_contract::migration::MigrationAppLabel::new(app_label.to_owned())
.map_err(display)?,
type_bridge_contract::migration::MigrationName::new(name.to_owned())
.map_err(display)?,
);
let manifest = graph.manifest(&id).ok_or_else(|| {
format!("approval target {compound:?} is not in the committed history")
})?;
type_bridge_schema_migration::MigrationApplyApproval::for_manifest(manifest)
.map_err(display)
})
.collect()
}
#[cfg(test)]
mod transport_option_tests {
use super::*;
fn environment(policy: WorkspaceTransportPolicy) -> WorkspaceEnvironment {
WorkspaceEnvironment::new(
"typedb.example:1729",
"example",
SecretReference::environment("TYPEBRIDGE_TEST_USERNAME").expect("username reference"),
SecretReference::environment("TYPEBRIDGE_TEST_PASSWORD").expect("password reference"),
)
.expect("environment")
.with_transport_policy(policy)
}
fn custom_root_workspace(root_bytes: &[u8]) -> (tempfile::TempDir, TypeBridgeWorkspace) {
let directory = tempfile::tempdir().expect("workspace directory");
fs::create_dir_all(directory.path().join("schema/fragments")).expect("schema directory");
fs::create_dir_all(directory.path().join("migrations/v2")).expect("migration directory");
fs::create_dir_all(directory.path().join("certs")).expect("certificate directory");
fs::write(
directory.path().join("schema/schema.yaml"),
"format: typebridge.schema-set/v1\nsources: [fragments/*.yaml]\n",
)
.expect("schema set writes");
fs::write(
directory.path().join("schema/fragments/model.yaml"),
"format: typebridge.schema/v2\nentities: {person: {}}\n",
)
.expect("schema writes");
fs::write(directory.path().join("certs/root.pem"), root_bytes).expect("certificate writes");
let manifest = directory.path().join("typebridge.yaml");
fs::write(
&manifest,
"format: typebridge.workspace/v1\n\
schema:\n root: schema/schema.yaml\n ownership: exclusive\n managed-scope: tls-test\n\
compatibility:\n semantic-profile: typedb-3.12.1/v1\n\
migrations:\n directory: migrations/v2\n app-label: tlstest\n\
environments:\n dev:\n database: tls_test\n uri: never-contact.invalid:1729\n \
tls: 'true'\n tls-root-ca: certs/root.pem\n credential:\n username: \
env:TYPEBRIDGE_TEST_USERNAME\n password: env:TYPEBRIDGE_TEST_PASSWORD\n",
)
.expect("manifest writes");
let workspace = load_workspace(&manifest).expect("custom-root workspace loads");
(directory, workspace)
}
#[test]
fn workspace_transport_policy_maps_without_changing_plaintext_defaults() {
let defaults = type_bridge_orm::SecureConnectOptions::default();
let disabled = secure_connect_options(&environment(WorkspaceTransportPolicy::Disabled));
assert_eq!(disabled.tls_mode, type_bridge_orm::TlsMode::Disabled);
assert_eq!(disabled.http_port, defaults.http_port);
assert_eq!(disabled.server_version, defaults.server_version);
let native = secure_connect_options(
&environment(WorkspaceTransportPolicy::NativeRoots).with_http_port(9443),
);
assert_eq!(native.tls_mode, type_bridge_orm::TlsMode::NativeRoots);
assert_eq!(native.http_port, 9443);
assert_eq!(native.server_version, defaults.server_version);
}
#[test]
fn custom_root_mapping_preserves_the_validated_canonical_path() {
let directory = tempfile::tempdir().expect("workspace directory");
let canonical = fs::canonicalize(directory.path()).expect("canonical workspace");
fs::create_dir_all(canonical.join("certs")).expect("certificate directory");
fs::write(
canonical.join("certs/root.pem"),
b"not parsed at workspace boundary\n",
)
.expect("certificate writes");
let root = WorkspaceRoot::new(canonical.clone()).expect("workspace root");
let root_ca = type_bridge_workspace::WorkspaceRootCa::new(
&root,
"certs/root.pem",
&SystemSchemaSourceService,
)
.expect("confined root CA");
let options = secure_connect_options(
&environment(WorkspaceTransportPolicy::CustomRootCa(root_ca)).with_http_port(8443),
);
assert_eq!(
options.tls_mode,
type_bridge_orm::TlsMode::CustomRootCa(canonical.join("certs/root.pem"))
);
assert_eq!(options.http_port, 8443);
}
#[test]
fn malformed_custom_root_fails_transport_preflight_before_credentials_are_needed() {
let (_directory, workspace) = custom_root_workspace(b"definitely not a certificate\n");
let error = preflight_secure_connect_options(&workspace, "dev")
.expect_err("PEM parsing must happen before credential resolution");
assert!(error.contains("tls_custom_root_ca_invalid_pem"), "{error}");
assert!(!error.contains("TYPEBRIDGE_TEST_USERNAME"), "{error}");
assert!(!error.contains("TYPEBRIDGE_TEST_PASSWORD"), "{error}");
}
#[cfg(unix)]
#[test]
fn workspace_root_swap_to_outside_symlink_is_rejected_at_transport_preflight() {
use std::os::unix::fs::symlink;
let (directory, workspace) = custom_root_workspace(b"initial regular root\n");
let outside = tempfile::tempdir().expect("outside directory");
let configured = directory.path().join("certs/root.pem");
let outside_root = outside.path().join("malicious.pem");
fs::write(
&outside_root,
include_bytes!("../../core/tests/fixtures/valid-root.pem"),
)
.expect("write outside replacement root");
fs::remove_file(&configured).expect("remove validated confined root");
symlink(&outside_root, &configured).expect("install outside symlink after validation");
let error = preflight_secure_connect_options(&workspace, "dev")
.expect_err("retained workspace paths must never follow a replacement symlink");
assert!(error.contains("tls_custom_root_ca_unreadable"), "{error}");
assert!(!error.contains("tls_custom_root_ca_invalid_pem"), "{error}");
}
#[cfg(unix)]
#[test]
fn real_directory_root_replacement_cannot_substitute_custom_trust() {
let (directory, workspace) =
custom_root_workspace(include_bytes!("../../core/tests/fixtures/valid-root.pem"));
let configured_root = directory.path().to_path_buf();
let held_root = configured_root.with_extension("retained-custom-root-ca");
fs::rename(&configured_root, &held_root).expect("move retained workspace root");
fs::create_dir_all(configured_root.join("certs")).expect("replacement root creates");
fs::write(
configured_root.join("certs/root.pem"),
b"attacker-controlled replacement is not a certificate\n",
)
.expect("replacement root writes");
let preflight = preflight_secure_connect_options(&workspace, "dev");
fs::remove_dir_all(&configured_root).expect("replacement root removes");
fs::rename(&held_root, &configured_root).expect("retained root restores");
preflight.expect("transport must use the CA under the retained original root");
}
}
#[cfg(test)]
mod adoption_file_tests {
use super::*;
use sha2::{Digest as _, Sha256};
const LEGACY_SCHEMA: &str = "define\nentity person;\n";
fn adoption_workspace() -> (tempfile::TempDir, TypeBridgeWorkspace) {
let directory = tempfile::tempdir().expect("workspace directory");
fs::create_dir_all(directory.path().join("schema/fragments")).expect("schema directory");
fs::write(
directory.path().join("schema/schema.yaml"),
"format: typebridge.schema-set/v1\nsources: [fragments/*.yaml]\n",
)
.expect("schema set writes");
fs::write(
directory.path().join("schema/fragments/model.yaml"),
"format: typebridge.schema/v2\nentities: {person: {}}\n",
)
.expect("schema writes");
let manifest = directory.path().join("typebridge.yaml");
fs::write(
&manifest,
"format: typebridge.workspace/v1\n\
schema:\n root: schema/schema.yaml\n ownership: exclusive\n managed-scope: adoption-test\n\
compatibility:\n semantic-profile: typedb-3.12.1/v1\n\
migrations:\n directory: migrations/v2\n app-label: smoke\n\
environments:\n dev:\n database: adoption_test\n uri: never-contact.invalid:1729\n migrate: 'true'\n credential:\n username: env:TYPEBRIDGE_TEST_USERNAME\n password: env:TYPEBRIDGE_TEST_PASSWORD\n",
)
.expect("manifest writes");
let workspace = load_workspace(&manifest).expect("workspace loads");
(directory, workspace)
}
fn write_legacy_fixture(root: &Path) -> PathBuf {
let directory = root.join("migrations/legacy");
fs::create_dir_all(&directory).expect("legacy directory");
let name = "0001_initial";
let python_source = "class Migration:\n operations = []\n";
let checksum = type_bridge_migration::migration_file_checksum(python_source);
let source_sha256 = format!("{:x}", Sha256::digest(python_source.as_bytes()));
let schema_hash = format!("{:x}", Sha256::digest(LEGACY_SCHEMA.as_bytes()));
fs::write(directory.join(format!("{name}.py")), python_source)
.expect("legacy source writes");
let adoption = type_bridge_migration::LegacyAdoptionMetadata::new(
"legacy",
name,
Vec::new(),
checksum,
source_sha256,
type_bridge_migration::LegacySchemaEffect::Snapshot,
type_bridge_migration::MigrationDependencySpec {
app_label: "legacy".to_owned(),
migration_name: name.to_owned(),
},
schema_hash.clone(),
)
.expect("legacy adoption metadata");
fs::write(
directory.join(format!("{name}.adoption.json")),
serde_json::to_vec_pretty(&adoption).expect("metadata encodes"),
)
.expect("metadata writes");
let snapshot = directory.join("snapshots/v0001");
fs::create_dir_all(&snapshot).expect("snapshot directory");
fs::write(snapshot.join("schema.tql"), LEGACY_SCHEMA).expect("snapshot schema writes");
fs::write(
snapshot.join("snapshot.json"),
serde_json::to_vec_pretty(&serde_json::json!({
"version": "v0001",
"source_migration": name,
"schema_hash": schema_hash,
"file_hashes": {"schema.tql": schema_hash},
"type_bridge_version": "1.5.11",
"type_bridge_core_version": "1.5.11"
}))
.expect("snapshot manifest encodes"),
)
.expect("snapshot manifest writes");
directory
}
#[test]
fn authority_publication_is_atomic_no_replace_and_resumable() {
let directory = tempfile::tempdir().expect("directory");
let authority =
type_bridge_schema_migration::MigrationDirectory::open_ambient(directory.path())
.expect("directory authority");
let name = "adopted-genesis.typeql";
let path = directory.path().join("adopted-genesis.typeql");
assert!(
publish_authority(&authority, name, b"define\nentity person;\n")
.expect("first publish")
);
assert!(
!publish_authority(&authority, name, b"define\nentity person;\n")
.expect("identical recovery")
);
assert!(publish_authority(&authority, name, b"define\nentity company;\n").is_err());
assert_eq!(
fs::read(&path).expect("authority reads"),
b"define\nentity person;\n"
);
assert!(
fs::read_dir(directory.path())
.expect("directory reads")
.all(|entry| !entry
.expect("entry")
.file_name()
.to_string_lossy()
.ends_with(".tmp"))
);
}
#[cfg(unix)]
#[test]
fn authority_publication_rejects_final_symlink() {
use std::os::unix::fs::symlink;
let directory = tempfile::tempdir().expect("directory");
let outside = directory.path().join("outside");
fs::write(&outside, b"untouched").expect("outside writes");
let path = directory.path().join("adopted-genesis.typeql");
symlink(&outside, &path).expect("symlink");
let authority =
type_bridge_schema_migration::MigrationDirectory::open_ambient(directory.path())
.expect("directory authority");
assert!(publish_authority(&authority, "adopted-genesis.typeql", b"replacement").is_err());
assert_eq!(fs::read(&outside).expect("outside reads"), b"untouched");
}
#[test]
fn invalid_adoption_name_creates_no_canonical_directory() {
let (directory, workspace) = adoption_workspace();
let missing_legacy = directory.path().join("missing-legacy");
let error = run_connected(
&workspace,
"dev",
ConnectedAction::Adopt {
legacy_directory: missing_legacy,
name: String::new(),
},
)
.expect_err("invalid name fails before history or network access");
assert!(error.contains("migration"), "{error}");
assert!(
!directory.path().join("migrations/v2").exists(),
"bad-name validation must not create canonical filesystem state"
);
}
#[test]
fn invalid_apply_approvals_fail_before_credentials_or_network() {
let (directory, workspace) = adoption_workspace();
fs::create_dir_all(directory.path().join("migrations/v2")).expect("canonical directory");
for (approval, expected) in [
("not-a-compound-id", "must be app-label/name"),
("smoke/0001_missing", "is not in the committed history"),
] {
let error = run_connected(
&workspace,
"dev",
ConnectedAction::Apply {
approvals: vec![approval.to_owned()],
},
)
.expect_err("invalid approval is rejected by local authority");
assert!(error.contains(expected), "{approval}: {error}");
assert!(
!error.contains("credential")
&& !error.contains("connect")
&& !error.contains("database"),
"approval validation ran after external setup: {error}"
);
}
}
#[test]
fn adoption_retry_completes_either_exact_orphan_direction() {
for orphan in ["genesis", "bridge"] {
let (directory, workspace) = adoption_workspace();
let legacy = write_legacy_fixture(directory.path());
let prepared = prepare_legacy_adoption(&workspace, &legacy, "0000_legacy_frontier")
.expect("adoption prepares");
let migration_directory = workspace
.ensure_migration_directory()
.expect("canonical directory");
match orphan {
"genesis" => {
publish_authority(
migration_directory.directory(),
type_bridge_schema_compat::ADOPTED_GENESIS_FILE_NAME,
prepared.reconstructed.schema_typeql().as_bytes(),
)
.expect("genesis orphan publishes");
}
"bridge" => {
publish_authority(
migration_directory.directory(),
&prepared.bridge_name,
&prepared.bridge_bytes,
)
.expect("bridge orphan publishes");
}
_ => unreachable!(),
}
publish_prepared_adoption(&workspace, &migration_directory, &prepared)
.expect("adoption retry completes the exact orphan");
workspace
.discover_migrations_in(&migration_directory)
.expect("completed adoption pair discovers");
assert!(
migration_directory
.display_path()
.join(type_bridge_schema_compat::ADOPTED_GENESIS_FILE_NAME)
.is_file()
);
assert!(
migration_directory
.display_path()
.join(&prepared.bridge_name)
.is_file()
);
}
}
#[test]
fn legacy_history_race_after_bridge_rolls_back_new_publication() {
let (directory, workspace) = adoption_workspace();
let legacy = write_legacy_fixture(directory.path());
let prepared = prepare_legacy_adoption(&workspace, &legacy, "0000_legacy_frontier")
.expect("adoption prepares");
let migration_directory = workspace
.ensure_migration_directory()
.expect("canonical directory");
let legacy_source = legacy.join("0001_initial.py");
let error = publish_prepared_adoption_with_after_bridge(
&workspace,
&migration_directory,
&prepared,
|| {
fs::write(
&legacy_source,
"class Migration:\n operations = ['changed']\n",
)
.expect("race mutation writes");
},
)
.expect_err("legacy authority race aborts pair publication");
assert!(error.contains("changed"), "{error}");
assert!(
!migration_directory
.display_path()
.join(&prepared.bridge_name)
.exists(),
"new bridge is rolled back"
);
assert!(
!migration_directory
.display_path()
.join(type_bridge_schema_compat::ADOPTED_GENESIS_FILE_NAME)
.exists(),
"genesis is never published after the race"
);
}
}