# WCO-K08: Configuration migration and versioning
**Area:** Widget Customization Docs
**Model:** moonshotai/kimi-k2.5
**Tokens:** 8,350
**Cost:** $0.0089
**Time:** 1841ms
**Timestamp:** 2026-02-16T13:59:28.154329

---

I'll build a comprehensive configuration versioning and migration system for Sunaiva. This includes versioned Pydantic schemas, a migration framework with dry-run support, PostgreSQL-backed history, diff visualization, and rollback capabilities.

```python
# config_versioning/__init__.py
"""Configuration versioning and migration system for Sunaiva Talking Widget."""

__version__ = "1.0.0"
```

```python
# config_versioning/models/schemas.py
"""Versioned configuration schemas with backward compatibility."""

from __future__ import annotations

from datetime import datetime
from enum import Enum
from typing import Any, Dict, List, Literal, Optional, Union

from pydantic import BaseModel, Field, validator, root_validator


class VoiceProvider(str, Enum):
    """Supported voice synthesis providers."""
    TELNYX = "telnyx"
    ELEVENLABS = "elevenlabs"
    AWS_POLLY = "aws_polly"
    GOOGLE_TTS = "google_tts"


class WidgetTheme(str, Enum):
    """Visual themes for the widget."""
    LIGHT = "light"
    DARK = "dark"
    AUTO = "auto"
    TRADIE_ORANGE = "tradie_orange"


# ============================================================================
# VERSION 1 SCHEMA (Legacy)
# ============================================================================

class V1WidgetConfig(BaseModel):
    """Version 1 configuration schema - Initial release."""
    version: Literal[1] = Field(1, description="Schema version")
    
    # Basic settings
    business_name: str = Field(..., min_length=1, max_length=100)
    phone_number: str = Field(..., regex=r'^\+61[0-9]{9}$')  # Australian format
    
    # Voice settings (simplified in v1)
    voice_id: str = Field("default", max_length=50)
    welcome_message: str = "Welcome to our business!"
    
    # Appearance
    primary_color: str = Field("#FF6B00", regex=r'^#[0-9A-Fa-f]{6}$')
    position: Literal["bottom-right", "bottom-left"] = "bottom-right"
    
    # Basic hours
    business_hours_start: str = "09:00"
    business_hours_end: str = "17:00"
    
    class Config:
        extra = "forbid"


# ============================================================================
# VERSION 2 SCHEMA (Current)
# ============================================================================

class V2BusinessHours(BaseModel):
    """Structured business hours for v2."""
    day: Literal["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
    start: str = Field(..., regex=r'^([01]?[0-9]|2[0-3]):[0-5][0-9]$')
    end: str = Field(..., regex=r'^([01]?[0-9]|2[0-3]):[0-5][0-9]$')
    closed: bool = False

    @root_validator
    def validate_hours(cls, values):
        if not values.get('closed'):
            start = values.get('start')
            end = values.get('end')
            if start and end:
                start_dt = datetime.strptime(start, "%H:%M")
                end_dt = datetime.strptime(end, "%H:%M")
                if start_dt >= end_dt:
                    raise ValueError("End time must be after start time")
        return values


class V2VoiceSettings(BaseModel):
    """Enhanced voice settings for v2."""
    provider: VoiceProvider = VoiceProvider.TELNYX
    voice_id: str = Field("default", max_length=50)
    welcome_message: str = Field("Welcome! How can I help you today?", max_length=500)
    language: str = Field("en-AU", regex=r'^[a-z]{2}-[A-Z]{2}$')
    speed: float = Field(1.0, ge=0.5, le=2.0)
    pitch: float = Field(1.0, ge=0.5, le=2.0)


class V2WidgetConfig(BaseModel):
    """Version 2 configuration schema - Enhanced voice and scheduling."""
    version: Literal[2] = Field(2, description="Schema version")
    migrated_at: Optional[datetime] = None
    
    # Business info
    business_name: str = Field(..., min_length=1, max_length=100)
    abn: Optional[str] = Field(None, regex=r'^\d{11}$')  # Australian Business Number
    phone_number: str = Field(..., regex=r'^\+61[0-9]{9}$')
    
    # Voice settings (structured)
    voice: V2VoiceSettings = Field(default_factory=V2VoiceSettings)
    
    # Appearance (enhanced)
    theme: WidgetTheme = WidgetTheme.TRADIE_ORANGE
    primary_color: str = Field("#FF6B00", regex=r'^#[0-9A-Fa-f]{6}$')
    secondary_color: str = Field("#2C3E50", regex=r'^#[0-9A-Fa-f]{6}$')
    position: Literal["bottom-right", "bottom-left", "top-right", "top-left"] = "bottom-right"
    custom_css: Optional[str] = None
    
    # Advanced scheduling
    business_hours: List[V2BusinessHours] = Field(
        default_factory=lambda: [
            V2BusinessHours(day="monday", start="09:00", end="17:00"),
            V2BusinessHours(day="tuesday", start="09:00", end="17:00"),
            V2BusinessHours(day="wednesday", start="09:00", end="17:00"),
            V2BusinessHours(day="thursday", start="09:00", end="17:00"),
            V2BusinessHours(day="friday", start="09:00", end="17:00"),
            V2BusinessHours(day="saturday", closed=True),
            V2BusinessHours(day="sunday", closed=True),
        ]
    )
    
    # Behavior
    auto_open_delay: int = Field(5, ge=0, le=60)  # seconds
    collect_lead_info: bool = True
    required_fields: List[str] = Field(default_factory=lambda: ["name", "phone"])
    
    class Config:
        extra = "forbid"


# ============================================================================
# VERSION 3 SCHEMA (Latest)
# ============================================================================

class V3AISettings(BaseModel):
    """AI/LLM configuration for v3."""
    provider: Literal["openai", "anthropic", "azure", "local"] = "openai"
    model: str = "gpt-4"
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(150, ge=50, le=1000)
    system_prompt: str = Field(
        "You are a helpful assistant for an Australian tradie business.",
        max_length=2000
    )
    knowledge_base_id: Optional[str] = None
    enable_rag: bool = True
    context_window: int = Field(10, ge=1, le=50)  # Previous messages to include


class V3AnalyticsSettings(BaseModel):
    """Analytics and tracking configuration."""
    enabled: bool = True
    anonymize_ip: bool = True
    track_conversions: bool = True
    google_analytics_id: Optional[str] = None
    facebook_pixel_id: Optional[str] = None


class V3CallHandling(BaseModel):
    """Advanced call routing and handling."""
    max_call_duration: int = Field(600, ge=60, le=3600)  # seconds
    voicemail_enabled: bool = True
    voicemail_message: str = "Please leave a message after the tone."
    call_forwarding_number: Optional[str] = Field(None, regex=r'^\+61[0-9]{9}$')
    overflow_after: int = Field(30, ge=10, le=300)  # seconds
    sms_notifications: bool = True
    notification_number: Optional[str] = Field(None, regex=r'^\+61[0-9]{9}$')


class V3WidgetConfig(BaseModel):
    """Version 3 configuration schema - AI integration and advanced routing."""
    version: Literal[3] = Field(3, description="Schema version")
    migrated_at: Optional[datetime] = None
    
    # Business info
    business_name: str = Field(..., min_length=1, max_length=100)
    abn: Optional[str] = Field(None, regex=r'^\d{11}$')
    phone_number: str = Field(..., regex=r'^\+61[0-9]{9}$')
    email: Optional[str] = None
    
    # Voice settings
    voice: V2VoiceSettings = Field(default_factory=V2VoiceSettings)
    
    # AI settings (new in v3)
    ai: V3AISettings = Field(default_factory=V3AISettings)
    
    # Appearance
    theme: WidgetTheme = WidgetTheme.TRADIE_ORANGE
    primary_color: str = Field("#FF6B00", regex=r'^#[0-9A-Fa-f]{6}$')
    secondary_color: str = Field("#2C3E50", regex=r'^#[0-9A-Fa-f]{6}$')
    position: Literal["bottom-right", "bottom-left", "top-right", "top-left"] = "bottom-right"
    custom_css: Optional[str] = None
    mobile_breakpoint: int = Field(768, ge=320, le=1920)
    
    # Scheduling
    business_hours: List[V2BusinessHours] = Field(default_factory=list)
    timezone: str = "Australia/Sydney"
    
    # Behavior
    auto_open_delay: int = Field(5, ge=0, le=60)
    collect_lead_info: bool = True
    required_fields: List[str] = Field(default_factory=lambda: ["name", "phone"])
    gdpr_compliant: bool = True
    privacy_policy_url: Optional[str] = None
    
    # Call handling (new in v3)
    call_handling: V3CallHandling = Field(default_factory=V3CallHandling)
    
    # Analytics (new in v3)
    analytics: V3AnalyticsSettings = Field(default_factory=V3AnalyticsSettings)
    
    # Feature flags
    features: Dict[str, bool] = Field(default_factory=dict)
    
    class Config:
        extra = "forbid"


# Union type for any config version
AnyConfig = Union[V1WidgetConfig, V2WidgetConfig, V3WidgetConfig]
LatestConfig = V3WidgetConfig
```

```python
# config_versioning/migrations/base.py
"""Base migration framework."""

from abc import ABC, abstractmethod
from typing import Any, Dict, Generic, Optional, TypeVar

from pydantic import BaseModel

T = TypeVar('T', bound=BaseModel)
U = TypeVar('U', bound=BaseModel)


class MigrationError(Exception):
    """Raised when a migration fails."""
    pass


class RollbackError(Exception):
    """Raised when a rollback fails."""
    pass


class MigrationResult(BaseModel):
    """Result of a migration operation."""
    success: bool
    version_from: int
    version_to: int
    data: Dict[str, Any]
    changes: Dict[str, Any]
    warnings: list[str] = []
    dry_run: bool = False
    
    class Config:
        arbitrary_types_allowed = True


class BaseMigration(ABC, Generic[T, U]):
    """Abstract base class for configuration migrations."""
    
    def __init__(self):
        self.changes: Dict[str, Any] = {}
        self.warnings: list[str] = []
    
    @property
    @abstractmethod
    def from_version(self) -> int:
        """Source version number."""
        pass
    
    @property
    @abstractmethod
    def to_version(self) -> int:
        """Target version number."""
        pass
    
    @abstractmethod
    def migrate(self, data: T) -> U:
        """Migrate from source to target version."""
        pass
    
    @abstractmethod
    def rollback(self, data: U) -> T:
        """Rollback from target to source version."""
        pass
    
    def validate_source(self, data: Dict[str, Any]) -> T:
        """Validate source data before migration."""
        raise NotImplementedError("Subclasses must implement validate_source")
    
    def validate_target(self, data: Dict[str, Any]) -> U:
        """Validate target data after migration."""
        raise NotImplementedError("Subclasses must implement validate_target")
    
    def execute(self, data: Dict[str, Any], dry_run: bool = False) -> MigrationResult:
        """
        Execute migration with validation and error handling.
        
        Args:
            data: Configuration data to migrate
            dry_run: If True, simulate migration without side effects
            
        Returns:
            MigrationResult with details of the operation
        """
        try:
            # Validate source
            source_obj = self.validate_source(data)
            
            # Perform migration
            if dry_run:
                # Deep copy for dry run simulation
                import copy
                test_data = copy.deepcopy(data)
                source_copy = self.validate_source(test_data)
                target_obj = self.migrate(source_copy)
                validated_target = self.validate_target(target_obj.dict())
            else:
                target_obj = self.migrate(source_obj)
                validated_target = self.validate_target(target_obj.dict())
            
            return MigrationResult(
                success=True,
                version_from=self.from_version,
                version_to=self.to_version,
                data=validated_target.dict() if isinstance(validated_target, BaseModel) else validated_target,
                changes=self.changes,
                warnings=self.warnings,
                dry_run=dry_run
            )
            
        except Exception as e:
            raise MigrationError(
                f"Migration from v{self.from_version} to v{self.to_version} failed: {str(e)}"
            ) from e
    
    def execute_rollback(self, data: Dict[str, Any], dry_run: bool = False) -> MigrationResult:
        """
        Execute rollback with validation.
        
        Args:
            data: Configuration data to rollback
            dry_run: If True, simulate rollback without side effects
            
        Returns:
            MigrationResult with details of the operation
        """
        try:
            # Validate target (current state)
            target_obj = self.validate_target(data)
            
            # Perform rollback
            if dry_run:
                import copy
                test_data = copy.deepcopy(data)
                target_copy = self.validate_target(test_data)
                source_obj = self.rollback(target_copy)
                validated_source = self.validate_source(source_obj.dict())
            else:
                source_obj = self.rollback(target_obj)
                validated_source = self.validate_source(source_obj.dict())
            
            return MigrationResult(
                success=True,
                version_from=self.to_version,
                version_to=self.from_version,
                data=validated_source.dict() if isinstance(validated_source, BaseModel) else validated_source,
                changes=self.changes,
                warnings=self.warnings,
                dry_run=dry_run
            )
            
        except Exception as e:
            raise RollbackError(
                f"Rollback from v{self.to_version} to v{self.from_version} failed: {str(e)}"
            ) from e
    
    def _add_change(self, path: str, old_value: Any, new_value: Any) -> None:
        """Record a change made during migration."""
        self.changes[path] = {
            "old": old_value,
            "new": new_value,
            "type": "modified" if old_value != new_value else "unchanged"
        }
    
    def _add_warning(self, message: str) -> None:
        """Add a warning about migration behavior."""
        self.warnings.append(message)
```

```python
# config_versioning/migrations/v1_to_v2.py
"""Migration from Version 1 to Version 2."""

from typing import Any, Dict

from config_versioning.models.schemas import V1WidgetConfig, V2BusinessHours, V2VoiceSettings, V2WidgetConfig
from config_versioning.migrations.base import BaseMigration


class V1ToV2Migration(BaseMigration[V1WidgetConfig, V2WidgetConfig]):
    """Migrate configuration from v1 to v2."""
    
    @property
    def from_version(self) -> int:
        return 1
    
    @property
    def to_version(self) -> int:
        return 2
    
    def validate_source(self, data: Dict[str, Any]) -> V1WidgetConfig:
        """Validate v1 configuration."""
        return V1WidgetConfig(**data)
    
    def validate_target(self, data: Dict[str, Any]) -> V2WidgetConfig:
        """Validate v2 configuration."""
        return V2WidgetConfig(**data)
    
    def migrate(self, data: V1WidgetConfig) -> V2WidgetConfig:
        """
        Migrate v1 config to v2.
        
        Changes:
        - Flat voice_id/welcome_message -> structured V2VoiceSettings
        - Simple business hours -> structured weekly schedule
        - Added ABN field
        - Added secondary_color
        - Added auto_open_delay
        - Added lead collection settings
        """
        from datetime import datetime
        
        # Map old flat voice settings to new structure
        voice_settings = V2VoiceSettings(
            provider="telnyx",  # Default for v1 users
            voice_id=data.voice_id,
            welcome_message=data.welcome_message,
            language="en-AU",
            speed=1.0,
            pitch=1.0
        )
        self._add_change("voice", "flat voice_id/welcome_message", "structured V2VoiceSettings")
        
        # Convert simple hours to structured weekly schedule
        business_hours = self._convert_business_hours(
            data.business_hours_start,
            data.business_hours_end
        )
        self._add_change(
            "business_hours", 
            f"{data.business_hours_start}-{data.business_hours_end} (daily)",
            "structured weekly schedule"
        )
        
        # Build v2 config
        v2_config = V2WidgetConfig(
            version=2,
            migrated_at=datetime.utcnow(),
            business_name=data.business_name,
            abn=None,  # New field, requires manual input
            phone_number=data.phone_number,
            voice=voice_settings,
            theme="tradie_orange",  # New default for v2
            primary_color=data.primary_color,
            secondary_color="#2C3E50",  # New default
            position=data.position,
            business_hours=business_hours,
            auto_open_delay=5,  # New default
            collect_lead_info=True,  # New default
            required_fields=["name", "phone"]  # New default
        )
        
        # Record new fields
        self._add_change("abn", None, "new field - requires manual input")
        self._add_change("secondary_color", None, "#2C3E50")
        self._add_change("auto_open_delay", None, 5)
        self._add_change("collect_lead_info", None, True)
        
        self._add_warning("ABN field added but set to None - please update manually")
        self._add_warning("Voice provider defaulted to 'telnyx' - verify in settings")
        
        return v2_config
    
    def rollback(self, data: V2WidgetConfig) -> V1WidgetConfig:
        """
        Rollback v2 config to v1.
        
        Lossy operation: structured data flattened, new fields discarded.
        """
        # Determine primary business hours (weekday)
        weekday_hours = next(
            (h for h in data.business_hours if h.day == "monday" and not h.closed),
            None
        )
        
        if not weekday_hours:
            # Fallback to first available hours
            weekday_hours = next(
                (h for h in data.business_hours if not h.closed),
                V2BusinessHours(day="monday", start="09:00", end="17:00")
            )
            self._add_warning("No Monday hours found, using first available day")
        
        v1_config = V1WidgetConfig(
            version=1,
            business_name=data.business_name,
            phone_number=data.phone_number,
            voice_id=data.voice.voice_id,
            welcome_message=data.voice.welcome_message,
            primary_color=data.primary_color,
            position=data.position if data.position in ["bottom-right", "bottom-left"] else "bottom-right",
            business_hours_start=weekday_hours.start,
            business_hours_end=weekday_hours.end
        )
        
        self._add_warning("Rollback is lossy: structured hours flattened to single schedule")
        self._add_warning("Secondary color, ABN, auto-open delay, and lead settings discarded")
        
        return v1_config
    
    def _convert_business_hours(self, start: str, end: str) -> list[V2BusinessHours]:
        """Convert v1 daily hours to v2 weekly structure."""
        weekdays = ["monday", "tuesday", "wednesday", "thursday", "friday"]
        weekend = ["saturday", "sunday"]
        
        hours = []
        
        # Weekdays
        for day in weekdays:
            hours.append(V2BusinessHours(day=day, start=start, end=end, closed=False))
        
        # Weekend closed by default
        for day in weekend:
            hours.append(V2BusinessHours(day=day, closed=True))
        
        return hours
```

```python
# config_versioning/migrations/v2_to_v3.py
"""Migration from Version 2 to Version 3."""

from typing import Any, Dict

from config_versioning.models.schemas import (
    V2WidgetConfig, 
    V3AISettings, 
    V3AnalyticsSettings, 
    V3CallHandling, 
    V3WidgetConfig
)
from config_versioning.migrations.base import BaseMigration


class V2ToV3Migration(BaseMigration[V2WidgetConfig, V3WidgetConfig]):
    """Migrate configuration from v2 to v3."""
    
    @property
    def from_version(self) -> int:
        return 2
    
    @property
    def to_version(self) -> int:
        return 3
    
    def validate_source(self, data: Dict[str, Any]) -> V2WidgetConfig:
        """Validate v2 configuration."""
        return V2WidgetConfig(**data)
    
    def validate_target(self, data: Dict[str, Any]) -> V3WidgetConfig:
        """Validate v3 configuration."""
        return V3WidgetConfig(**data)
    
    def migrate(self, data: V2WidgetConfig) -> V3WidgetConfig:
        """
        Migrate v2 config to v3.
        
        Changes:
        - Added AI settings (OpenAI integration)
        - Added call handling configuration
        - Added analytics settings
        - Added GDPR compliance flags
        - Added timezone support
        - Added mobile breakpoint
        """
        from datetime import datetime
        
        # Initialize new AI settings with defaults
        ai_settings = V3AISettings(
            provider="openai",
            model="gpt-4",
            temperature=0.7,
            max_tokens=150,
            system_prompt=f"You are a helpful assistant for {data.business_name}.",
            enable_rag=False,  # Disabled by default, requires setup
            context_window=10
        )
        self._add_change("ai", None, "new AI settings block")
        
        # Initialize call handling
        call_handling = V3CallHandling(
            max_call_duration=600,
            voicemail_enabled=True,
            voicemail_message="Please leave a message after the tone.",
            call_forwarding_number=data.phone_number,  # Default to business number
            overflow_after=30,
            sms_notifications=False,  # Requires opt-in
            notification_number=data.phone_number
        )
        self._add_change("call_handling", None, "new call handling block")
        
        # Initialize analytics
        analytics = V3AnalyticsSettings(
            enabled=True,
            anonymize_ip=True,
            track_conversions=True
        )
        self._add_change("analytics", None, "new analytics block")
        
        # Build v3 config
        v3_config = V3WidgetConfig(
            version=3,
            migrated_at=datetime.utcnow(),
            business_name=data.business_name,
            abn=data.abn,
            phone_number=data.phone_number,
            email=None,  # New field
            voice=data.voice,
            ai=ai_settings,
            theme=data.theme,
            primary_color=data.primary_color,
            secondary_color=data.secondary_color,
            position=data.position,
            custom_css=data.custom_css,
            mobile_breakpoint=768,  # New default
            business_hours=data.business_hours,
            timezone="Australia/Sydney",  # Default for Australian tradies
            auto_open_delay=data.auto_open_delay,
            collect_lead_info=data.collect_lead_info,
            required_fields=data.required_fields,
            gdpr_compliant=True,  # New default
            privacy_policy_url=None,  # New field
            call_handling=call_handling,
            analytics=analytics,
            features={}  # New feature flags
        )
        
        self._add_change("email", None, "new field")
        self._add_change("mobile_breakpoint", None, 768)
        self._add_change("timezone", None, "Australia/Sydney")
        self._add_change("gdpr_compliant", None, True)
        self._add_change("privacy_policy_url", None, "new field")
        self._add_change("features", None, {})
        
        self._add_warning("AI settings initialized with defaults - review and configure")
        self._add_warning("Call forwarding defaulted to business number - verify settings")
        self._add_warning("GDPR compliance enabled by default - ensure privacy policy is set")
        
        return v3_config
    
    def rollback(self, data: V3WidgetConfig) -> V2WidgetConfig:
        """
        Rollback v3 config to v2.
        
        Lossy operation: AI settings, call handling, analytics removed.
        """
        from datetime import datetime
        
        v2_config = V2WidgetConfig(
            version=2,
            migrated_at=datetime.utcnow(),
            business_name=data.business_name,
            abn=data.abn,
            phone_number=data.phone_number,
            voice=data.voice,
            theme=data.theme,
            primary_color=data.primary_color,
            secondary_color=data.secondary_color,
            position=data.position,
            custom_css=data.custom_css,
            business_hours=data.business_hours,
            auto_open_delay=data.auto_open_delay,
            collect_lead_info=data.collect_lead_info,
            required_fields=data.required_fields
        )
        
        self._add_warning("Rollback is lossy: AI settings, call handling, and analytics removed")
        self._add_warning("GDPR settings and timezone information lost")
        self._add_warning("Email and mobile breakpoint settings discarded")
        
        return v2_config
```

```python
# config_versioning/migrations/registry.py
"""Migration registry and chain execution."""

from typing import Dict, List, Optional, Type

from config_versioning.migrations.base import BaseMigration, MigrationError, MigrationResult


class MigrationRegistry:
    """Registry for managing available migrations."""
    
    def __init__(self):
        self._migrations: Dict[tuple[int, int], BaseMigration] = {}
    
    def register(self, migration: BaseMigration) -> None:
        """Register a migration."""
        key = (migration.from_version, migration.to_version)
        if key in self._migrations:
            raise MigrationError(f"Migration from v{key[0]} to v{key[1]} already registered")
        self._migrations[key] = migration
    
    def get_migration(self, from_version: int, to_version: int) -> Optional[BaseMigration]:
        """Get specific migration."""
        return self._migrations.get((from_version, to_version))
    
    def find_migration_path(self, from_version: int, to_version: int) -> List[BaseMigration]:
        """
        Find migration path using BFS.
        
        Returns ordered list of migrations to execute.
        """
        if from_version == to_version:
            return []
        
        if from_version < to_version:
            # Upgrade path
            return self._find_upgrade_path(from_version, to_version)
        else:
            # Downgrade path
            return self._find_downgrade_path(from_version, to_version)
    
    def _find_upgrade_path(self, start: int, target: int) -> List[BaseMigration]:
        """Find upgrade path using BFS."""
        from collections import deque
        
        queue = deque([(start, [])])
        visited = {start}
        
        while queue:
            current_version, path = queue.popleft()
            
            if current_version == target:
                return path
            
            # Find all possible next steps
            for (fv, tv), migration in self._migrations.items():
                if fv == current_version and tv not in visited:
                    visited.add(tv)
                    queue.append((tv, path + [migration]))
        
        raise MigrationError(f"No migration path found from v{start} to v{target}")
    
    def _find_downgrade_path(self, start: int, target: int) -> List[BaseMigration]:
        """Find downgrade path using BFS."""
        from collections import deque
        
        queue = deque([(start, [])])
        visited = {start}
        
        while queue:
            current_version, path = queue.popleft()
            
            if current_version == target:
                return path
            
            # Find all possible rollbacks (reverse migrations)
            for (fv, tv), migration in self._migrations.items():
                if tv == current_version and fv not in visited:
                    visited.add(fv)
                    queue.append((fv, path + [migration]))
        
        raise MigrationError(f"No rollback path found from v{start} to v{target}")


class MigrationChain:
    """Execute chained migrations."""
    
    def __init__(self, registry: MigrationRegistry):
        self.registry = registry
    
    def migrate(
        self, 
        data: Dict[str, Any], 
        target_version: int,
        dry_run: bool = False
    ) -> List[MigrationResult]:
        """
        Execute migration chain to target version.
        
        Args:
            data: Current configuration
            target_version: Desired version
            dry_run: If True, simulate without saving
            
        Returns:
            List of migration results
        """
        current_version = data.get("version", 1)
        
        if current_version == target_version:
            return []
        
        migrations = self.registry.find_migration_path(current_version, target_version)
        results = []
        current_data = data
        
        for migration in migrations:
            result = migration.execute(current_data, dry_run=dry_run)
            results.append(result)
            current_data = result.data
            
            if not result.success:
                raise MigrationError(
                    f"Migration failed at v{migration.from_version} -> v{migration.to_version}"
                )
        
        return results
    
    def rollback(
        self,
        data: Dict[str, Any],
        target_version: int,
        dry_run: bool = False
    ) -> List[MigrationResult]:
        """
        Execute rollback chain to target version.
        
        Args:
            data: Current configuration
            target_version: Desired version
            dry_run: If True, simulate without saving
            
        Returns:
            List of rollback results
        """
        current_version = data.get("version", 1)
        
        if current_version == target_version:
            return []
        
        migrations = self.registry.find_migration_path(current_version, target_version)
        results = []
        current_data = data
        
        for migration in migrations:
            result = migration.execute_rollback(current_data, dry_run=dry_run)
            results.append(result)
            current_data = result.data
            
            if not result.success:
                raise MigrationError(
                    f"Rollback failed at v{migration.to_version} -> v{migration.from_version}"
                )
        
        return results
```

```python
# config_versioning/storage/database.py
"""Database storage for configuration history."""

from datetime import datetime
from typing import Any, Dict, List, Optional
from uuid import UUID, uuid4

import asyncpg
from pydantic import BaseModel


class ConfigHistoryEntry(BaseModel):
    """Configuration history record."""
    id: UUID
    business_id: str
    version: int
    config_data: Dict[str, Any]
    migrated_from: Optional[int] = None
    migrated_to: Optional[int] = None
    migration_changes: Optional[Dict[str, Any]] = None
    created_by: Optional[str] = None
    created_at: datetime
    is_rollback: bool = False
    rollback_to_id: Optional[UUID] = None
    
    class Config:
        arbitrary_types_allowed = True


class ConfigStorage:
    """PostgreSQL storage for configuration versioning."""
    
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool
    
    async def initialize(self):
        """Create necessary tables."""
        async with self.pool.acquire() as conn:
            await conn.execute("""
                CREATE TABLE IF NOT EXISTS config_history (
                    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
                    business_id VARCHAR(255) NOT NULL,
                    version INTEGER NOT NULL,
                    config_data JSONB NOT NULL,
                    migrated_from INTEGER,
                    migrated_to INTEGER,
                    migration_changes JSONB,
                    created_by VARCHAR(255),
                    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
                    is_rollback BOOLEAN DEFAULT FALSE,
                    rollback_to_id UUID REFERENCES config_history(id),
                    CONSTRAINT valid_rollback CHECK (
                        (is_rollback = FALSE AND rollback_to_id IS NULL) OR
                        (is_rollback = TRUE AND rollback_to_id IS NOT NULL)
                    )
                );
                
                CREATE INDEX IF NOT EXISTS idx_config_history_business 
                    ON config_history(business_id, created_at DESC);
                
                CREATE INDEX IF NOT EXISTS idx_config_history_version 
                    ON config_history(business_id, version);
            """)
    
    async def save_config(
        self,
        business_id: str,
        config: Dict[str, Any],
        created_by: Optional[str] = None,
        migrated_from: Optional[int] = None,
        migrated_to: Optional[int] = None,
        migration_changes: Optional[Dict[str, Any]] = None,
        is_rollback: bool = False,
        rollback_to_id: Optional[UUID] = None
    ) -> ConfigHistoryEntry:
        """Save configuration snapshot."""
        query = """
            INSERT INTO config_history 
                (business_id, version, config_data, migrated_from, migrated_to, 
                 migration_changes, created_by, is_rollback, rollback_to_id)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            RETURNING id, business_id, version, config_data, migrated_from, migrated_to,
                      migration_changes, created_by, created_at, is_rollback, rollback_to_id
        """
        
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                query,
                business_id,
                config.get("version", 1),
                config,
                migrated_from,
                migrated_to,
                migration_changes,
                created_by,
                is_rollback,
                rollback_to_id
            )
            
            return ConfigHistoryEntry(**dict(row))
    
    async def get_current_config(self, business_id: str) -> Optional[Dict[str, Any]]:
        """Get latest configuration for business."""
        query = """
            SELECT config_data 
            FROM config_history 
            WHERE business_id = $1 
            ORDER BY created_at DESC 
            LIMIT 1
        """
        
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(query, business_id)
            return row["config_data"] if row else None
    
    async def get_config_at_version(
        self, 
        business_id: str, 
        version: int
    ) -> Optional[Dict[str, Any]]:
        """Get configuration at specific version."""
        query = """
            SELECT config_data 
            FROM config_history 
            WHERE business_id = $1 AND version = $2
            ORDER BY created_at DESC 
            LIMIT 1
        """
        
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(query,