882 lines
30 KiB
Rust
882 lines
30 KiB
Rust
//! Feature versioning and lineage tracking
|
|
//!
|
|
//! This module provides comprehensive version management for features
|
|
//! including lineage tracking, schema evolution, and backward compatibility.
|
|
|
|
use chrono::{DateTime, Utc};
|
|
use serde::{Deserialize, Serialize};
|
|
use std::collections::HashMap;
|
|
use std::sync::Arc;
|
|
use tokio::sync::RwLock;
|
|
use uuid::Uuid;
|
|
|
|
use crate::{FeatureStoreError, Result};
|
|
|
|
/// Feature version identifier
|
|
pub type FeatureVersion = Uuid;
|
|
|
|
/// Feature version metadata
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct VersionMetadata {
|
|
/// Version identifier
|
|
pub version_id: FeatureVersion,
|
|
/// Feature name this version belongs to
|
|
pub feature_name: String,
|
|
/// Version number (semantic versioning style)
|
|
pub version_number: String,
|
|
/// Schema for this version
|
|
pub schema: serde_json::Value,
|
|
/// Description of changes in this version
|
|
pub description: String,
|
|
/// Author of this version
|
|
pub author: String,
|
|
/// Creation timestamp
|
|
pub created_at: DateTime<Utc>,
|
|
/// Parent version (for tracking evolution)
|
|
pub parent_version: Option<FeatureVersion>,
|
|
/// Whether this version is deprecated
|
|
pub deprecated: bool,
|
|
/// Deprecation timestamp
|
|
pub deprecated_at: Option<DateTime<Utc>>,
|
|
/// Migration path to newer version
|
|
pub migration_path: Option<String>,
|
|
/// Tags for categorization
|
|
pub tags: Vec<String>,
|
|
}
|
|
|
|
/// Feature lineage information
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct FeatureLineage {
|
|
/// Feature name
|
|
pub feature_name: String,
|
|
/// All versions of this feature in chronological order
|
|
pub versions: Vec<VersionMetadata>,
|
|
/// Features that this feature depends on
|
|
pub dependencies: Vec<FeatureDependency>,
|
|
/// Creation timestamp of the lineage
|
|
pub created_at: DateTime<Utc>,
|
|
}
|
|
|
|
/// Feature dependency information
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct FeatureDependency {
|
|
/// Name of the dependent feature
|
|
pub feature_name: String,
|
|
/// Version of the dependent feature
|
|
pub version_id: Option<FeatureVersion>,
|
|
/// Type of dependency
|
|
pub dependency_type: DependencyType,
|
|
/// When this dependency was established
|
|
pub created_at: DateTime<Utc>,
|
|
}
|
|
|
|
/// Types of feature dependencies
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum DependencyType {
|
|
/// Direct dependency - feature is computed from this dependency
|
|
Direct,
|
|
/// Indirect dependency - feature transitively depends on this
|
|
Indirect,
|
|
/// Schema dependency - feature schema is derived from this
|
|
Schema,
|
|
/// Temporal dependency - feature requires historical values from this
|
|
Temporal,
|
|
}
|
|
|
|
/// Schema compatibility result
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct CompatibilityResult {
|
|
/// Whether schemas are compatible
|
|
pub compatible: bool,
|
|
/// Level of compatibility
|
|
pub compatibility_level: CompatibilityLevel,
|
|
/// Issues found during compatibility check
|
|
pub issues: Vec<CompatibilityIssue>,
|
|
/// Suggested migration steps
|
|
pub migration_suggestions: Vec<String>,
|
|
}
|
|
|
|
/// Levels of schema compatibility
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum CompatibilityLevel {
|
|
/// Fully backward compatible
|
|
FullyCompatible,
|
|
/// Backward compatible with minor issues
|
|
BackwardCompatible,
|
|
/// Forward compatible only
|
|
ForwardCompatible,
|
|
/// Incompatible - breaking changes
|
|
Incompatible,
|
|
}
|
|
|
|
/// Compatibility issue description
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct CompatibilityIssue {
|
|
/// Type of issue
|
|
pub issue_type: IssueType,
|
|
/// Description of the issue
|
|
pub description: String,
|
|
/// Path in schema where issue occurs
|
|
pub schema_path: String,
|
|
/// Severity of the issue
|
|
pub severity: IssueSeverity,
|
|
}
|
|
|
|
/// Types of compatibility issues
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum IssueType {
|
|
/// Field was removed
|
|
FieldRemoved,
|
|
/// Field type changed
|
|
TypeChanged,
|
|
/// Required field added
|
|
RequiredFieldAdded,
|
|
/// Constraint tightened
|
|
ConstraintTightened,
|
|
/// Format changed
|
|
FormatChanged,
|
|
}
|
|
|
|
/// Issue severity levels
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum IssueSeverity {
|
|
/// Breaking change - requires migration
|
|
Breaking,
|
|
/// Warning - might cause issues
|
|
Warning,
|
|
/// Info - FYI but not problematic
|
|
Info,
|
|
}
|
|
|
|
/// Version manager for handling feature versions
|
|
pub struct VersionManager {
|
|
/// Version metadata storage
|
|
versions: Arc<RwLock<HashMap<FeatureVersion, VersionMetadata>>>,
|
|
/// Feature to current version mapping
|
|
current_versions: Arc<RwLock<HashMap<String, FeatureVersion>>>,
|
|
/// Lineage information
|
|
lineages: Arc<RwLock<HashMap<String, FeatureLineage>>>,
|
|
}
|
|
|
|
impl VersionManager {
|
|
/// Create a new version manager
|
|
#[must_use]
|
|
pub fn new() -> Self {
|
|
Self {
|
|
versions: Arc::new(RwLock::new(HashMap::new())),
|
|
current_versions: Arc::new(RwLock::new(HashMap::new())),
|
|
lineages: Arc::new(RwLock::new(HashMap::new())),
|
|
}
|
|
}
|
|
|
|
/// Create a new version of a feature
|
|
pub async fn create_version(
|
|
&self,
|
|
feature_name: &str,
|
|
schema: serde_json::Value,
|
|
description: &str,
|
|
author: &str,
|
|
parent_version: Option<FeatureVersion>,
|
|
) -> Result<FeatureVersion> {
|
|
let version_id = Uuid::new_v4();
|
|
|
|
// Determine version number
|
|
let version_number = self
|
|
.determine_version_number(feature_name, parent_version)
|
|
.await?;
|
|
|
|
let metadata = VersionMetadata {
|
|
version_id,
|
|
feature_name: feature_name.to_string(),
|
|
version_number,
|
|
schema,
|
|
description: description.to_string(),
|
|
author: author.to_string(),
|
|
created_at: Utc::now(),
|
|
parent_version,
|
|
deprecated: false,
|
|
deprecated_at: None,
|
|
migration_path: None,
|
|
tags: vec![],
|
|
};
|
|
|
|
// Store version metadata
|
|
{
|
|
let mut versions = self.versions.write().await;
|
|
versions.insert(version_id, metadata.clone());
|
|
}
|
|
|
|
// Update current version
|
|
{
|
|
let mut current_versions = self.current_versions.write().await;
|
|
current_versions.insert(feature_name.to_string(), version_id);
|
|
}
|
|
|
|
// Update lineage
|
|
self.update_lineage(feature_name, metadata).await?;
|
|
|
|
Ok(version_id)
|
|
}
|
|
|
|
/// Get version metadata
|
|
pub async fn get_version(&self, version_id: FeatureVersion) -> Result<VersionMetadata> {
|
|
let versions = self.versions.read().await;
|
|
versions
|
|
.get(&version_id)
|
|
.cloned()
|
|
.ok_or_else(|| FeatureStoreError::NotFound(format!("Version {version_id} not found")))
|
|
}
|
|
|
|
/// Get current version of a feature
|
|
pub async fn get_current_version(&self, feature_name: &str) -> Result<FeatureVersion> {
|
|
let current_versions = self.current_versions.read().await;
|
|
current_versions.get(feature_name).copied().ok_or_else(|| {
|
|
FeatureStoreError::NotFound(format!("No current version for feature {feature_name}"))
|
|
})
|
|
}
|
|
|
|
/// Get all versions of a feature
|
|
pub async fn get_feature_versions(&self, feature_name: &str) -> Result<Vec<VersionMetadata>> {
|
|
let versions = self.versions.read().await;
|
|
let feature_versions: Vec<VersionMetadata> = versions
|
|
.values()
|
|
.filter(|v| v.feature_name == feature_name)
|
|
.cloned()
|
|
.collect();
|
|
|
|
if feature_versions.is_empty() {
|
|
Err(FeatureStoreError::NotFound(format!(
|
|
"No versions found for feature {feature_name}"
|
|
)))
|
|
} else {
|
|
Ok(feature_versions)
|
|
}
|
|
}
|
|
|
|
/// Get feature lineage
|
|
pub async fn get_lineage(&self, feature_name: &str) -> Result<FeatureLineage> {
|
|
let lineages = self.lineages.read().await;
|
|
lineages.get(feature_name).cloned().ok_or_else(|| {
|
|
FeatureStoreError::NotFound(format!("Lineage for feature {feature_name} not found"))
|
|
})
|
|
}
|
|
|
|
/// Add dependency between features
|
|
pub async fn add_dependency(
|
|
&self,
|
|
feature_name: &str,
|
|
dependency_name: &str,
|
|
dependency_type: DependencyType,
|
|
dependency_version: Option<FeatureVersion>,
|
|
) -> Result<()> {
|
|
let dependency = FeatureDependency {
|
|
feature_name: dependency_name.to_string(),
|
|
version_id: dependency_version,
|
|
dependency_type,
|
|
created_at: Utc::now(),
|
|
};
|
|
|
|
let mut lineages = self.lineages.write().await;
|
|
if let Some(lineage) = lineages.get_mut(feature_name) {
|
|
lineage.dependencies.push(dependency);
|
|
} else {
|
|
return Err(FeatureStoreError::NotFound(format!(
|
|
"Lineage for feature {feature_name} not found"
|
|
)));
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Check schema compatibility between two versions
|
|
pub async fn check_compatibility(
|
|
&self,
|
|
from_version: FeatureVersion,
|
|
to_version: FeatureVersion,
|
|
) -> Result<CompatibilityResult> {
|
|
let versions = self.versions.read().await;
|
|
|
|
let from_metadata = versions.get(&from_version).ok_or_else(|| {
|
|
FeatureStoreError::NotFound(format!("Version {from_version} not found"))
|
|
})?;
|
|
|
|
let to_metadata = versions.get(&to_version).ok_or_else(|| {
|
|
FeatureStoreError::NotFound(format!("Version {to_version} not found"))
|
|
})?;
|
|
|
|
self.analyze_schema_compatibility(&from_metadata.schema, &to_metadata.schema)
|
|
}
|
|
|
|
/// Deprecate a version
|
|
pub async fn deprecate_version(
|
|
&self,
|
|
version_id: FeatureVersion,
|
|
migration_path: Option<String>,
|
|
) -> Result<()> {
|
|
let mut versions = self.versions.write().await;
|
|
if let Some(metadata) = versions.get_mut(&version_id) {
|
|
metadata.deprecated = true;
|
|
metadata.deprecated_at = Some(Utc::now());
|
|
metadata.migration_path = migration_path;
|
|
Ok(())
|
|
} else {
|
|
Err(FeatureStoreError::NotFound(format!(
|
|
"Version {version_id} not found"
|
|
)))
|
|
}
|
|
}
|
|
|
|
/// Get deprecated versions
|
|
pub async fn get_deprecated_versions(&self) -> Vec<VersionMetadata> {
|
|
let versions = self.versions.read().await;
|
|
versions
|
|
.values()
|
|
.filter(|v| v.deprecated)
|
|
.cloned()
|
|
.collect()
|
|
}
|
|
|
|
/// Generate version migration plan
|
|
pub async fn generate_migration_plan(
|
|
&self,
|
|
from_version: FeatureVersion,
|
|
to_version: FeatureVersion,
|
|
) -> Result<MigrationPlan> {
|
|
let compatibility = self.check_compatibility(from_version, to_version).await?;
|
|
|
|
let steps = match compatibility.compatibility_level {
|
|
CompatibilityLevel::FullyCompatible => vec![MigrationStep {
|
|
step_type: MigrationStepType::NoAction,
|
|
description: "No migration required - fully compatible".to_string(),
|
|
sql_query: None,
|
|
validation: None,
|
|
}],
|
|
CompatibilityLevel::BackwardCompatible => vec![MigrationStep {
|
|
step_type: MigrationStepType::SchemaUpdate,
|
|
description: "Update schema with backward compatible changes".to_string(),
|
|
sql_query: Some("ALTER TABLE feature_metadata UPDATE schema...".to_string()),
|
|
validation: Some("Validate all existing data still conforms".to_string()),
|
|
}],
|
|
CompatibilityLevel::ForwardCompatible => vec![MigrationStep {
|
|
step_type: MigrationStepType::DataTransformation,
|
|
description: "Transform data to match new schema".to_string(),
|
|
sql_query: Some(
|
|
"UPDATE feature_values SET value = transform(value)...".to_string(),
|
|
),
|
|
validation: Some("Validate transformed data integrity".to_string()),
|
|
}],
|
|
CompatibilityLevel::Incompatible => {
|
|
let mut steps = vec![MigrationStep {
|
|
step_type: MigrationStepType::Backup,
|
|
description: "Backup existing data before migration".to_string(),
|
|
sql_query: Some(
|
|
"CREATE TABLE feature_values_backup AS SELECT * FROM feature_values..."
|
|
.to_string(),
|
|
),
|
|
validation: Some("Verify backup completeness".to_string()),
|
|
}];
|
|
|
|
for issue in &compatibility.issues {
|
|
match issue.issue_type {
|
|
IssueType::FieldRemoved => {
|
|
steps.push(MigrationStep {
|
|
step_type: MigrationStepType::DataTransformation,
|
|
description: format!("Remove field: {}", issue.schema_path),
|
|
sql_query: Some(format!(
|
|
"UPDATE feature_values SET value = json_remove(value, '{}')...",
|
|
issue.schema_path
|
|
)),
|
|
validation: Some(
|
|
"Verify field removal doesn't break dependencies".to_string(),
|
|
),
|
|
});
|
|
}
|
|
IssueType::TypeChanged => {
|
|
steps.push(MigrationStep {
|
|
step_type: MigrationStepType::DataTransformation,
|
|
description: format!("Convert type for field: {}", issue.schema_path),
|
|
sql_query: Some(format!("UPDATE feature_values SET value = convert_type(value, '{}')...", issue.schema_path)),
|
|
validation: Some("Verify type conversion preserves semantics".to_string()),
|
|
});
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
steps.push(MigrationStep {
|
|
step_type: MigrationStepType::SchemaUpdate,
|
|
description: "Update schema to new version".to_string(),
|
|
sql_query: Some(
|
|
"UPDATE feature_metadata SET schema = $1 WHERE name = $2".to_string(),
|
|
),
|
|
validation: Some("Verify all data conforms to new schema".to_string()),
|
|
});
|
|
|
|
steps
|
|
}
|
|
};
|
|
|
|
Ok(MigrationPlan {
|
|
from_version,
|
|
to_version,
|
|
compatibility_level: compatibility.compatibility_level,
|
|
estimated_duration_secs: steps.len() as u64 * 60, // Rough estimate in seconds
|
|
steps,
|
|
rollback_plan: self
|
|
.generate_rollback_plan(to_version, from_version)
|
|
.await?,
|
|
created_at: Utc::now(),
|
|
})
|
|
}
|
|
|
|
/// Execute version migration
|
|
pub async fn execute_migration(&self, plan: &MigrationPlan) -> Result<MigrationResult> {
|
|
let start_time = Utc::now();
|
|
let mut executed_steps = Vec::new();
|
|
let mut errors = Vec::new();
|
|
|
|
for (index, step) in plan.steps.iter().enumerate() {
|
|
let step_start = Utc::now();
|
|
|
|
match self.execute_migration_step(step).await {
|
|
Ok(()) => {
|
|
executed_steps.push(ExecutedStep {
|
|
step_index: index,
|
|
step_type: step.step_type.clone(),
|
|
started_at: step_start,
|
|
completed_at: Utc::now(),
|
|
success: true,
|
|
error: None,
|
|
});
|
|
}
|
|
Err(e) => {
|
|
errors.push(format!("Step {index}: {e}"));
|
|
executed_steps.push(ExecutedStep {
|
|
step_index: index,
|
|
step_type: step.step_type.clone(),
|
|
started_at: step_start,
|
|
completed_at: Utc::now(),
|
|
success: false,
|
|
error: Some(e.to_string()),
|
|
});
|
|
break; // Stop on first error
|
|
}
|
|
}
|
|
}
|
|
|
|
let success = errors.is_empty();
|
|
let end_time = Utc::now();
|
|
|
|
Ok(MigrationResult {
|
|
migration_id: Uuid::new_v4(),
|
|
from_version: plan.from_version,
|
|
to_version: plan.to_version,
|
|
started_at: start_time,
|
|
completed_at: end_time,
|
|
success,
|
|
executed_steps,
|
|
errors,
|
|
})
|
|
}
|
|
|
|
// Private helper methods
|
|
|
|
async fn determine_version_number(
|
|
&self,
|
|
_feature_name: &str,
|
|
parent_version: Option<FeatureVersion>,
|
|
) -> Result<String> {
|
|
if let Some(parent_id) = parent_version {
|
|
let versions = self.versions.read().await;
|
|
if let Some(parent) = versions.get(&parent_id) {
|
|
// Increment patch version
|
|
let parts: Vec<&str> = parent.version_number.split('.').collect();
|
|
if parts.len() == 3 {
|
|
let major: u32 = parts[0].parse().unwrap_or(1);
|
|
let minor: u32 = parts[1].parse().unwrap_or(0);
|
|
let patch: u32 = parts[2].parse().unwrap_or(0);
|
|
Ok(format!("{}.{}.{}", major, minor, patch + 1))
|
|
} else {
|
|
Ok("1.0.1".to_string())
|
|
}
|
|
} else {
|
|
Ok("1.0.0".to_string())
|
|
}
|
|
} else {
|
|
// First version
|
|
Ok("1.0.0".to_string())
|
|
}
|
|
}
|
|
|
|
async fn update_lineage(&self, feature_name: &str, metadata: VersionMetadata) -> Result<()> {
|
|
let mut lineages = self.lineages.write().await;
|
|
|
|
if let Some(lineage) = lineages.get_mut(feature_name) {
|
|
lineage.versions.push(metadata);
|
|
lineage
|
|
.versions
|
|
.sort_by(|a, b| a.created_at.cmp(&b.created_at));
|
|
} else {
|
|
let lineage = FeatureLineage {
|
|
feature_name: feature_name.to_string(),
|
|
versions: vec![metadata],
|
|
dependencies: vec![],
|
|
created_at: Utc::now(),
|
|
};
|
|
lineages.insert(feature_name.to_string(), lineage);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn analyze_schema_compatibility(
|
|
&self,
|
|
from_schema: &serde_json::Value,
|
|
to_schema: &serde_json::Value,
|
|
) -> Result<CompatibilityResult> {
|
|
let mut issues = Vec::new();
|
|
let mut migration_suggestions = Vec::new();
|
|
|
|
// Simple compatibility check - in a real implementation this would be much more sophisticated
|
|
if from_schema == to_schema {
|
|
return Ok(CompatibilityResult {
|
|
compatible: true,
|
|
compatibility_level: CompatibilityLevel::FullyCompatible,
|
|
issues: vec![],
|
|
migration_suggestions: vec![],
|
|
});
|
|
}
|
|
|
|
// Check for type changes
|
|
if let (Some(from_type), Some(to_type)) = (from_schema.get("type"), to_schema.get("type"))
|
|
&& from_type != to_type
|
|
{
|
|
issues.push(CompatibilityIssue {
|
|
issue_type: IssueType::TypeChanged,
|
|
description: format!("Type changed from {from_type} to {to_type}"),
|
|
schema_path: "type".to_string(),
|
|
severity: IssueSeverity::Breaking,
|
|
});
|
|
migration_suggestions.push("Data transformation required for type change".to_string());
|
|
}
|
|
|
|
// Check for property changes in objects
|
|
if let (Some(from_props), Some(to_props)) =
|
|
(from_schema.get("properties"), to_schema.get("properties"))
|
|
&& let (Some(from_obj), Some(to_obj)) = (from_props.as_object(), to_props.as_object())
|
|
{
|
|
// Check for removed properties
|
|
for (key, _) in from_obj {
|
|
if !to_obj.contains_key(key) {
|
|
issues.push(CompatibilityIssue {
|
|
issue_type: IssueType::FieldRemoved,
|
|
description: format!("Property '{key}' was removed"),
|
|
schema_path: format!("properties.{key}"),
|
|
severity: IssueSeverity::Breaking,
|
|
});
|
|
migration_suggestions
|
|
.push(format!("Remove property '{key}' from existing data"));
|
|
}
|
|
}
|
|
|
|
// Check for added required properties
|
|
if let Some(to_required) = to_schema.get("required")
|
|
&& let Some(required_array) = to_required.as_array()
|
|
{
|
|
for req_prop in required_array {
|
|
if let Some(prop_name) = req_prop.as_str()
|
|
&& !from_obj.contains_key(prop_name)
|
|
{
|
|
issues.push(CompatibilityIssue {
|
|
issue_type: IssueType::RequiredFieldAdded,
|
|
description: format!("Required property '{prop_name}' was added"),
|
|
schema_path: format!("properties.{prop_name}"),
|
|
severity: IssueSeverity::Breaking,
|
|
});
|
|
migration_suggestions.push(format!(
|
|
"Add default value for required property '{prop_name}'"
|
|
));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let compatibility_level = if issues
|
|
.iter()
|
|
.any(|i| matches!(i.severity, IssueSeverity::Breaking))
|
|
{
|
|
CompatibilityLevel::Incompatible
|
|
} else if issues
|
|
.iter()
|
|
.any(|i| matches!(i.severity, IssueSeverity::Warning))
|
|
{
|
|
CompatibilityLevel::BackwardCompatible
|
|
} else {
|
|
CompatibilityLevel::FullyCompatible
|
|
};
|
|
|
|
Ok(CompatibilityResult {
|
|
compatible: !matches!(compatibility_level, CompatibilityLevel::Incompatible),
|
|
compatibility_level,
|
|
issues,
|
|
migration_suggestions,
|
|
})
|
|
}
|
|
|
|
async fn generate_rollback_plan(
|
|
&self,
|
|
from_version: FeatureVersion,
|
|
to_version: FeatureVersion,
|
|
) -> Result<Vec<MigrationStep>> {
|
|
// Generate rollback plan (reverse of migration)
|
|
Ok(vec![MigrationStep {
|
|
step_type: MigrationStepType::Rollback,
|
|
description: format!("Rollback from version {from_version} to {to_version}"),
|
|
sql_query: Some("RESTORE FROM BACKUP...".to_string()),
|
|
validation: Some("Verify rollback completed successfully".to_string()),
|
|
}])
|
|
}
|
|
|
|
async fn execute_migration_step(&self, step: &MigrationStep) -> Result<()> {
|
|
// In a real implementation, this would execute the actual migration steps
|
|
tracing::info!("Executing migration step: {}", step.description);
|
|
|
|
match step.step_type {
|
|
MigrationStepType::NoAction => Ok(()),
|
|
MigrationStepType::Backup => {
|
|
// Simulate backup
|
|
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
|
|
Ok(())
|
|
}
|
|
MigrationStepType::SchemaUpdate => {
|
|
// Simulate schema update
|
|
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
|
|
Ok(())
|
|
}
|
|
MigrationStepType::DataTransformation => {
|
|
// Simulate data transformation
|
|
tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
|
|
Ok(())
|
|
}
|
|
MigrationStepType::Rollback => {
|
|
// Simulate rollback
|
|
tokio::time::sleep(tokio::time::Duration::from_millis(300)).await;
|
|
Ok(())
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Default for VersionManager {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
/// Migration plan for version transitions
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct MigrationPlan {
|
|
/// Source version
|
|
pub from_version: FeatureVersion,
|
|
/// Target version
|
|
pub to_version: FeatureVersion,
|
|
/// Compatibility level
|
|
pub compatibility_level: CompatibilityLevel,
|
|
/// Estimated duration for migration in seconds
|
|
pub estimated_duration_secs: u64,
|
|
/// Migration steps
|
|
pub steps: Vec<MigrationStep>,
|
|
/// Rollback plan
|
|
pub rollback_plan: Vec<MigrationStep>,
|
|
/// Plan creation timestamp
|
|
pub created_at: DateTime<Utc>,
|
|
}
|
|
|
|
/// Individual migration step
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct MigrationStep {
|
|
/// Type of migration step
|
|
pub step_type: MigrationStepType,
|
|
/// Description of the step
|
|
pub description: String,
|
|
/// Optional SQL query for the step
|
|
pub sql_query: Option<String>,
|
|
/// Optional validation query
|
|
pub validation: Option<String>,
|
|
}
|
|
|
|
/// Types of migration steps
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub enum MigrationStepType {
|
|
/// No action required
|
|
NoAction,
|
|
/// Backup existing data
|
|
Backup,
|
|
/// Update schema
|
|
SchemaUpdate,
|
|
/// Transform data
|
|
DataTransformation,
|
|
/// Rollback changes
|
|
Rollback,
|
|
}
|
|
|
|
/// Result of migration execution
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct MigrationResult {
|
|
/// Migration identifier
|
|
pub migration_id: Uuid,
|
|
/// Source version
|
|
pub from_version: FeatureVersion,
|
|
/// Target version
|
|
pub to_version: FeatureVersion,
|
|
/// Migration start time
|
|
pub started_at: DateTime<Utc>,
|
|
/// Migration completion time
|
|
pub completed_at: DateTime<Utc>,
|
|
/// Whether migration was successful
|
|
pub success: bool,
|
|
/// Executed steps
|
|
pub executed_steps: Vec<ExecutedStep>,
|
|
/// Any errors encountered
|
|
pub errors: Vec<String>,
|
|
}
|
|
|
|
/// Information about an executed migration step
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub struct ExecutedStep {
|
|
/// Step index in the plan
|
|
pub step_index: usize,
|
|
/// Type of step
|
|
pub step_type: MigrationStepType,
|
|
/// When step started
|
|
pub started_at: DateTime<Utc>,
|
|
/// When step completed
|
|
pub completed_at: DateTime<Utc>,
|
|
/// Whether step succeeded
|
|
pub success: bool,
|
|
/// Error message if failed
|
|
pub error: Option<String>,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn test_version_manager() {
|
|
let vm = VersionManager::new();
|
|
|
|
// Create initial version
|
|
let schema_v1 = serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"age": {"type": "integer"}
|
|
},
|
|
"required": ["name"]
|
|
});
|
|
|
|
let v1 = vm
|
|
.create_version(
|
|
"user_profile",
|
|
schema_v1,
|
|
"Initial version",
|
|
"test_author",
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(vm.get_current_version("user_profile").await.unwrap(), v1);
|
|
|
|
// Create second version
|
|
let schema_v2 = serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"age": {"type": "integer"},
|
|
"email": {"type": "string"}
|
|
},
|
|
"required": ["name", "email"]
|
|
});
|
|
|
|
let v2 = vm
|
|
.create_version(
|
|
"user_profile",
|
|
schema_v2,
|
|
"Added email field",
|
|
"test_author",
|
|
Some(v1),
|
|
)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert_eq!(vm.get_current_version("user_profile").await.unwrap(), v2);
|
|
|
|
// Check compatibility
|
|
let compatibility = vm.check_compatibility(v1, v2).await.unwrap();
|
|
assert!(!compatibility.compatible); // Should be incompatible due to new required field
|
|
assert!(matches!(
|
|
compatibility.compatibility_level,
|
|
CompatibilityLevel::Incompatible
|
|
));
|
|
|
|
// Get lineage
|
|
let lineage = vm.get_lineage("user_profile").await.unwrap();
|
|
assert_eq!(lineage.versions.len(), 2);
|
|
assert_eq!(lineage.versions[0].version_number, "1.0.0");
|
|
assert_eq!(lineage.versions[1].version_number, "1.0.1");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_migration_plan() {
|
|
let vm = VersionManager::new();
|
|
|
|
let schema_v1 = serde_json::json!({"type": "string"});
|
|
let v1 = vm
|
|
.create_version("test_feature", schema_v1, "v1", "author", None)
|
|
.await
|
|
.unwrap();
|
|
|
|
let schema_v2 = serde_json::json!({"type": "integer"});
|
|
let v2 = vm
|
|
.create_version("test_feature", schema_v2, "v2", "author", Some(v1))
|
|
.await
|
|
.unwrap();
|
|
|
|
let plan = vm.generate_migration_plan(v1, v2).await.unwrap();
|
|
assert!(!plan.steps.is_empty());
|
|
assert!(matches!(
|
|
plan.compatibility_level,
|
|
CompatibilityLevel::Incompatible
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn test_schema_compatibility() {
|
|
let vm = VersionManager::new();
|
|
|
|
let schema1 = serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"}
|
|
}
|
|
});
|
|
|
|
let schema2 = serde_json::json!({
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"optional_field": {"type": "integer"}
|
|
}
|
|
});
|
|
|
|
let result = vm.analyze_schema_compatibility(&schema1, &schema2).unwrap();
|
|
assert!(result.compatible);
|
|
assert!(matches!(
|
|
result.compatibility_level,
|
|
CompatibilityLevel::FullyCompatible
|
|
));
|
|
}
|
|
}
|