Deployment Strategies.

# deployment/blue_green.py

"""
Blue-Green deployment orchestration.

This manages the switch between environments
and provides instant rollback capability.
"""

from dataclasses import dataclass
from enum import Enum
from typing import Optional
from datetime import datetime, timezone
import asyncio
import logging
import uuid

logger = logging.getLogger(__name__)


class Environment(Enum):
    BLUE = "blue"
    GREEN = "green"


class DeploymentState(Enum):
    IDLE = "idle"
    DEPLOYING = "deploying"
    TESTING = "testing"
    SWITCHING = "switching"
    VERIFYING = "verifying"
    COMPLETE = "complete"
    ROLLED_BACK = "rolled_back"


@dataclass
class DeploymentRecord:
    """Record of a deployment."""
    id: str
    version: str
    started_at: datetime
    completed_at: Optional[datetime]
    state: DeploymentState
    active_environment: Environment
    previous_environment: Environment
    deployed_by: str
    rollback_reason: Optional[str] = None


class BlueGreenDeployer:
    """
    Orchestrates blue-green deployments.
    """
    
    def __init__(
        self,
        load_balancer,
        blue_cluster,
        green_cluster,
        health_checker,
        metrics_client
    ):
        self.lb = load_balancer
        self.blue = blue_cluster
        self.green = green_cluster
        self.health = health_checker
        self.metrics = metrics_client
        
        self.current_environment = None
        self.current_deployment = None
    
    async def deploy(
        self,
        version: str,
        deployed_by: str,
        skip_tests: bool = False
    ) -> DeploymentRecord:
        """
        Execute a blue-green deployment.
        """
        # Determine which environment to deploy to
        if self.current_environment == Environment.BLUE:
            target = Environment.GREEN
            target_cluster = self.green
        else:
            target = Environment.BLUE
            target_cluster = self.blue
        
        deployment = DeploymentRecord(
            id=str(uuid.uuid4()),
            version=version,
            started_at=datetime.now(timezone.utc),
            completed_at=None,
            state=DeploymentState.DEPLOYING,
            active_environment=self.current_environment,
            previous_environment=self.current_environment,
            deployed_by=deployed_by
        )
        
        try:
            # Step 1: Deploy to inactive environment
            logger.info(f"Deploying {version} to {target.value}")
            deployment.state = DeploymentState.DEPLOYING
            
            await target_cluster.deploy(version)
            
            # Step 2: Run health checks and tests
            logger.info(f"Running health checks on {target.value}")
            deployment.state = DeploymentState.TESTING
            
            if not skip_tests:
                health_result = await self.health.check_cluster(target_cluster)
                if not health_result.healthy:
                    raise DeploymentError(f"Health check failed: {health_result.reason}")
                
                # Run smoke tests
                test_result = await self._run_smoke_tests(target_cluster)
                if not test_result.passed:
                    raise DeploymentError(f"Smoke tests failed: {test_result.failures}")
            
            # Step 3: Switch traffic
            logger.info(f"Switching traffic to {target.value}")
            deployment.state = DeploymentState.SWITCHING
            
            await self.lb.switch_to(target)
            
            # Step 4: Verify in production
            logger.info("Verifying deployment")
            deployment.state = DeploymentState.VERIFYING
            
            # Wait and check metrics
            await asyncio.sleep(30)  # Give it 30 seconds
            
            verification = await self._verify_deployment(version)
            if not verification.success:
                logger.warning(f"Verification failed: {verification.reason}")
                await self.rollback(deployment, verification.reason)
                return deployment
            
            # Step 5: Complete
            deployment.state = DeploymentState.COMPLETE
            deployment.completed_at = datetime.now(timezone.utc)
            deployment.active_environment = target
            
            self.current_environment = target
            self.current_deployment = deployment
            
            logger.info(f"Deployment {deployment.id} complete")
            
            return deployment
            
        except Exception as e:
            logger.error(f"Deployment failed: {e}")
            await self.rollback(deployment, str(e))
            raise
    
    async def rollback(
        self,
        deployment: DeploymentRecord,
        reason: str
    ):
        """
        Rollback to previous environment.
        
        This is INSTANT because the old environment is still running.
        """
        logger.warning(f"Rolling back deployment {deployment.id}: {reason}")
        
        deployment.state = DeploymentState.ROLLED_BACK
        deployment.rollback_reason = reason
        deployment.completed_at = datetime.now(timezone.utc)
        
        # Switch traffic back to previous environment
        await self.lb.switch_to(deployment.previous_environment)
        
        # Record rollback metric
        self.metrics.increment(
            "deployments_rolled_back_total",
            labels={"version": deployment.version, "reason": reason}
        )
        
        logger.info(f"Rollback complete. Traffic restored to {deployment.previous_environment.value}")
    
    async def _run_smoke_tests(self, cluster) -> TestResult:
        """Run smoke tests against a cluster."""
        tests = [
            self._test_health_endpoint(cluster),
            self._test_api_endpoint(cluster),
            self._test_database_connectivity(cluster),
        ]
        
        results = await asyncio.gather(*tests, return_exceptions=True)
        failures = [r for r in results if isinstance(r, Exception) or not r.passed]
        
        return TestResult(
            passed=len(failures) == 0,
            failures=[str(f) for f in failures]
        )
    
    async def _verify_deployment(self, version: str) -> VerificationResult:
        """Verify deployment is healthy in production."""
        
        # Check error rate
        error_rate = await self.metrics.query(
            'sum(rate(http_requests_total{status=~"5.."}[1m])) / '
            'sum(rate(http_requests_total[1m]))'
        )
        
        if error_rate > 0.01:  # > 1% error rate
            return VerificationResult(
                success=False,
                reason=f"Error rate too high: {error_rate:.2%}"
            )
        
        # Check latency
        p99_latency = await self.metrics.query(
            'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[1m])) by (le))'
        )
        
        if p99_latency > 1.0:  # > 1 second
            return VerificationResult(
                success=False,
                reason=f"Latency too high: {p99_latency:.2f}s"
            )
        
        return VerificationResult(success=True, reason=None)

# deployment/canary.py

"""
Canary deployment with automated analysis.

Gradually rolls out new version while comparing
metrics against the stable version.
"""

from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime, timedelta, timezone
from enum import Enum
import asyncio
import logging

logger = logging.getLogger(__name__)


@dataclass
class CanaryStage:
    """A stage in the canary rollout."""
    name: str
    traffic_percentage: int
    duration_minutes: int
    auto_promote: bool = True  # Auto-promote if metrics pass


@dataclass
class CanaryConfig:
    """Configuration for canary deployment."""
    stages: List[CanaryStage] = field(default_factory=lambda: [
        CanaryStage("canary-1pct", 1, 15),
        CanaryStage("canary-10pct", 10, 15),
        CanaryStage("canary-50pct", 50, 15),
        CanaryStage("full-rollout", 100, 0),
    ])
    
    # Thresholds for automated analysis
    max_error_rate_increase: float = 0.01  # 1% absolute increase
    max_latency_increase_pct: float = 0.10  # 10% relative increase
    min_request_count: int = 100  # Minimum requests before analysis


@dataclass
class CanaryAnalysis:
    """Result of canary analysis."""
    passed: bool
    stable_error_rate: float
    canary_error_rate: float
    stable_p99_latency: float
    canary_p99_latency: float
    request_count: int
    reason: Optional[str] = None


class CanaryDeployer:
    """
    Orchestrates canary deployments with automated analysis.
    """
    
    def __init__(
        self,
        kubernetes_client,
        metrics_client,
        traffic_manager,
        config: CanaryConfig = None
    ):
        self.k8s = kubernetes_client
        self.metrics = metrics_client
        self.traffic = traffic_manager
        self.config = config or CanaryConfig()
    
    async def deploy(
        self,
        version: str,
        deployment_name: str,
        namespace: str = "default"
    ) -> CanaryResult:
        """
        Execute a canary deployment.
        """
        logger.info(f"Starting canary deployment of {version}")
        
        result = CanaryResult(
            version=version,
            started_at=datetime.now(timezone.utc),
            stages_completed=[]
        )
        
        try:
            # Deploy canary pods (initially with 0% traffic)
            await self._deploy_canary_pods(version, deployment_name, namespace)
            
            # Progress through stages
            for stage in self.config.stages:
                logger.info(f"Entering stage: {stage.name} ({stage.traffic_percentage}%)")
                
                # Shift traffic
                await self.traffic.set_canary_weight(
                    deployment_name,
                    stage.traffic_percentage
                )
                
                # Wait for stage duration
                if stage.duration_minutes > 0:
                    await asyncio.sleep(stage.duration_minutes * 60)
                    
                    # Analyze metrics
                    analysis = await self._analyze_canary(deployment_name)
                    
                    if not analysis.passed:
                        logger.warning(f"Canary failed at {stage.name}: {analysis.reason}")
                        await self._rollback(deployment_name, namespace)
                        result.failed = True
                        result.failure_reason = analysis.reason
                        result.failure_stage = stage.name
                        return result
                    
                    logger.info(f"Stage {stage.name} passed analysis")
                
                result.stages_completed.append(stage.name)
            
            # Promote canary to stable
            await self._promote_canary(deployment_name, namespace)
            
            result.completed_at = datetime.now(timezone.utc)
            result.success = True
            
            logger.info(f"Canary deployment complete: {version}")
            
            return result
            
        except Exception as e:
            logger.error(f"Canary deployment failed: {e}")
            await self._rollback(deployment_name, namespace)
            result.failed = True
            result.failure_reason = str(e)
            return result
    
    async def _analyze_canary(self, deployment_name: str) -> CanaryAnalysis:
        """
        Compare canary metrics against stable.
        """
        # Get error rates
        stable_errors = await self.metrics.query(f'''
            sum(rate(http_requests_total{{
                deployment="{deployment_name}",
                version="stable",
                status=~"5.."
            }}[5m])) /
            sum(rate(http_requests_total{{
                deployment="{deployment_name}",
                version="stable"
            }}[5m]))
        ''')
        
        canary_errors = await self.metrics.query(f'''
            sum(rate(http_requests_total{{
                deployment="{deployment_name}",
                version="canary",
                status=~"5.."
            }}[5m])) /
            sum(rate(http_requests_total{{
                deployment="{deployment_name}",
                version="canary"
            }}[5m]))
        ''')
        
        # Get latencies
        stable_latency = await self.metrics.query(f'''
            histogram_quantile(0.99,
                sum(rate(http_request_duration_seconds_bucket{{
                    deployment="{deployment_name}",
                    version="stable"
                }}[5m])) by (le)
            )
        ''')
        
        canary_latency = await self.metrics.query(f'''
            histogram_quantile(0.99,
                sum(rate(http_request_duration_seconds_bucket{{
                    deployment="{deployment_name}",
                    version="canary"
                }}[5m])) by (le)
            )
        ''')
        
        # Get request count
        request_count = await self.metrics.query(f'''
            sum(increase(http_requests_total{{
                deployment="{deployment_name}",
                version="canary"
            }}[5m]))
        ''')
        
        # Check thresholds
        analysis = CanaryAnalysis(
            passed=True,
            stable_error_rate=stable_errors,
            canary_error_rate=canary_errors,
            stable_p99_latency=stable_latency,
            canary_p99_latency=canary_latency,
            request_count=int(request_count)
        )
        
        # Not enough traffic to analyze: fail closed, never promote on no
        # evidence (alternatively, extend the stage and re-check)
        if analysis.request_count < self.config.min_request_count:
            logger.warning(f"Not enough requests for analysis: {analysis.request_count}")
            analysis.passed = False
            analysis.reason = (
                f"Insufficient canary traffic ({analysis.request_count} < "
                f"{self.config.min_request_count} requests)"
            )
            return analysis
        
        # Check error rate
        error_increase = canary_errors - stable_errors
        if error_increase > self.config.max_error_rate_increase:
            analysis.passed = False
            analysis.reason = (
                f"Error rate increased by {error_increase:.2%} "
                f"(threshold: {self.config.max_error_rate_increase:.2%})"
            )
            return analysis
        
        # Check latency
        if stable_latency > 0:
            latency_increase = (canary_latency - stable_latency) / stable_latency
            if latency_increase > self.config.max_latency_increase_pct:
                analysis.passed = False
                analysis.reason = (
                    f"Latency increased by {latency_increase:.2%} "
                    f"(threshold: {self.config.max_latency_increase_pct:.2%})"
                )
                return analysis
        
        return analysis
    
    async def _rollback(self, deployment_name: str, namespace: str):
        """Rollback canary deployment."""
        logger.info(f"Rolling back canary for {deployment_name}")
        
        # Remove canary traffic
        await self.traffic.set_canary_weight(deployment_name, 0)
        
        # Delete canary pods
        await self.k8s.delete_canary_deployment(deployment_name, namespace)
        
        logger.info("Rollback complete")
    
    async def _promote_canary(self, deployment_name: str, namespace: str):
        """Promote canary to stable."""
        logger.info(f"Promoting canary to stable for {deployment_name}")
        
        # Update stable deployment with canary version
        await self.k8s.promote_canary(deployment_name, namespace)
        
        # Remove canary traffic split
        await self.traffic.remove_canary_route(deployment_name)
        
        logger.info("Promotion complete")

# deployment/feature_flags.py

"""
Feature flag system for controlled rollouts.

Supports:
- Boolean flags (on/off)
- Percentage rollouts
- User targeting
- Environment targeting
"""

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Callable
from enum import Enum
import hashlib
import logging

logger = logging.getLogger(__name__)


class RolloutStrategy(Enum):
    ALL = "all"           # Flag applies to everyone
    NONE = "none"         # Flag applies to no one
    PERCENTAGE = "percentage"  # Random percentage of users
    USER_LIST = "user_list"    # Specific users
    ATTRIBUTE = "attribute"     # Based on user attributes


@dataclass
class FeatureFlag:
    """A feature flag definition."""
    key: str
    name: str
    description: str
    
    # Targeting
    strategy: RolloutStrategy
    percentage: int = 0  # For PERCENTAGE strategy
    user_list: List[str] = field(default_factory=list)  # For USER_LIST
    attribute_rules: List[Dict] = field(default_factory=list)  # For ATTRIBUTE
    
    # Metadata
    owner: str = ""
    created_at: datetime = None
    expires_at: Optional[datetime] = None
    
    # Default value
    default_value: bool = False
    
    # Variants (for A/B tests)
    variants: Dict[str, Any] = field(default_factory=dict)


@dataclass
class EvaluationContext:
    """Context for evaluating a feature flag."""
    user_id: Optional[str] = None
    tenant_id: Optional[str] = None
    environment: str = "production"
    attributes: Dict[str, Any] = field(default_factory=dict)


class FeatureFlagService:
    """
    Feature flag evaluation service.
    """
    
    def __init__(self, flag_store, metrics_client):
        self.store = flag_store
        self.metrics = metrics_client
        self._cache = {}
    
    def is_enabled(
        self,
        flag_key: str,
        context: EvaluationContext = None
    ) -> bool:
        """
        Check if a feature flag is enabled.
        
        Usage:
            if feature_flags.is_enabled("new_checkout", context):
                return new_checkout_flow()
            else:
                return old_checkout_flow()
        """
        context = context or EvaluationContext()
        
        try:
            flag = self._get_flag(flag_key)
            
            if flag is None:
                logger.warning(f"Unknown flag: {flag_key}")
                return False
            
            # Check if expired
            if flag.expires_at and datetime.now(timezone.utc) > flag.expires_at:
                return flag.default_value
            
            # Evaluate based on strategy
            result = self._evaluate(flag, context)
            
            # Record metric
            self.metrics.increment(
                "feature_flag_evaluation_total",
                labels={
                    "flag": flag_key,
                    "result": str(result),
                    "strategy": flag.strategy.value
                }
            )
            
            return result
            
        except Exception as e:
            logger.error(f"Error evaluating flag {flag_key}: {e}")
            return False
    
    def get_variant(
        self,
        flag_key: str,
        context: EvaluationContext = None
    ) -> Optional[str]:
        """
        Get variant for A/B test flags.
        
        Usage:
            variant = feature_flags.get_variant("button_color", context)
            # Returns "blue" or "green" based on user bucketing
        """
        context = context or EvaluationContext()
        flag = self._get_flag(flag_key)
        
        if flag is None or not flag.variants:
            return None
        
        # Consistent hashing to assign user to variant
        bucket = self._get_bucket(flag_key, context.user_id)
        
        # Distribute buckets across variants
        cumulative = 0
        for variant_name, percentage in flag.variants.items():
            cumulative += percentage
            if bucket < cumulative:
                return variant_name
        
        return list(flag.variants.keys())[0]  # Default to first variant
    
    def _evaluate(self, flag: FeatureFlag, context: EvaluationContext) -> bool:
        """Evaluate flag based on its strategy."""
        
        if flag.strategy == RolloutStrategy.ALL:
            return True
        
        if flag.strategy == RolloutStrategy.NONE:
            return False
        
        if flag.strategy == RolloutStrategy.PERCENTAGE:
            return self._evaluate_percentage(flag, context)
        
        if flag.strategy == RolloutStrategy.USER_LIST:
            return context.user_id in flag.user_list
        
        if flag.strategy == RolloutStrategy.ATTRIBUTE:
            return self._evaluate_attributes(flag, context)
        
        return flag.default_value
    
    def _evaluate_percentage(
        self,
        flag: FeatureFlag,
        context: EvaluationContext
    ) -> bool:
        """
        Percentage-based evaluation with consistent hashing.
        
        Same user always gets same result for same flag.
        """
        bucket = self._get_bucket(flag.key, context.user_id)
        return bucket < flag.percentage
    
    def _get_bucket(self, flag_key: str, user_id: str) -> int:
        """
        Get consistent bucket (0-99) for user and flag.
        
        Same user + flag = same bucket, always.
        """
        if not user_id:
            # Random for anonymous users (NOT sticky: pass a session or device ID for a stable assignment)
            import random
            return random.randint(0, 99)
        
        # Hash user_id + flag_key for consistent bucketing
        hash_input = f"{flag_key}:{user_id}"
        hash_value = hashlib.md5(hash_input.encode()).hexdigest()
        return int(hash_value[:8], 16) % 100
    
    def _evaluate_attributes(
        self,
        flag: FeatureFlag,
        context: EvaluationContext
    ) -> bool:
        """Evaluate attribute-based rules."""
        for rule in flag.attribute_rules:
            attribute = rule.get("attribute")
            operator = rule.get("operator")
            value = rule.get("value")
            
            user_value = context.attributes.get(attribute)
            
            if operator == "equals" and user_value == value:
                return True
            if operator == "in" and user_value in value:
                return True
            if operator == "greater_than" and user_value > value:
                return True
        
        return flag.default_value
    
    def _get_flag(self, flag_key: str) -> Optional[FeatureFlag]:
        """Get flag from store with caching.

        Note: this cache has no TTL, so flag changes (including kill switches)
        would not propagate until restart. Real systems expire or refresh entries.
        """
        if flag_key in self._cache:
            return self._cache[flag_key]
        
        flag = self.store.get(flag_key)
        if flag:
            self._cache[flag_key] = flag
        
        return flag


# =============================================================================
# USAGE EXAMPLES
# =============================================================================

# Initialize
feature_flags = FeatureFlagService(flag_store, metrics)

# Simple boolean check
if feature_flags.is_enabled("new_checkout"):
    process_with_new_checkout()
else:
    process_with_old_checkout()

# With user context
context = EvaluationContext(
    user_id="user-123",
    tenant_id="tenant-456",
    attributes={
        "plan": "enterprise",
        "country": "US"
    }
)

if feature_flags.is_enabled("advanced_analytics", context):
    show_advanced_analytics()

# A/B test
button_color = feature_flags.get_variant("button_color_test", context)
render_button(color=button_color)

# deployment/migrations.py

"""
Safe database migration patterns.

Implements expand-contract for breaking changes.
"""

from dataclasses import dataclass
from enum import Enum
from typing import List, Optional
import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)


class MigrationPhase(Enum):
    EXPAND = "expand"
    MIGRATE = "migrate"
    CONTRACT = "contract"


@dataclass
class Migration:
    """A database migration."""
    version: str
    name: str
    phase: MigrationPhase
    
    # SQL to run
    up_sql: str
    down_sql: str
    
    # Validation
    pre_check_sql: Optional[str] = None  # Must pass before running
    post_check_sql: Optional[str] = None  # Must pass after running
    
    # Safety
    requires_lock: bool = False  # Requires table lock
    estimated_duration: str = ""  # e.g., "5 minutes", "2 hours"
    affects_writes: bool = False  # Will block writes


class SafeMigrationRunner:
    """
    Runs migrations safely with validation and rollback support.
    """
    
    def __init__(self, db_connection, lock_manager):
        self.db = db_connection
        self.locks = lock_manager
    
    async def run_migration(
        self,
        migration: Migration,
        dry_run: bool = False
    ) -> MigrationResult:
        """
        Run a migration with safety checks.
        """
        logger.info(f"Starting migration: {migration.name}")
        
        result = MigrationResult(
            migration=migration,
            started_at=datetime.now(timezone.utc)
        )
        
        try:
            # Pre-check
            if migration.pre_check_sql:
                check_result = await self.db.fetch_one(migration.pre_check_sql)
                if not check_result or not check_result[0]:
                    raise MigrationError("Pre-check failed")
            
            # Acquire lock if needed
            if migration.requires_lock:
                await self.locks.acquire("migration_lock")
            
            # Run migration
            if not dry_run:
                await self.db.execute(migration.up_sql)
            else:
                logger.info(f"DRY RUN: Would execute: {migration.up_sql}")
            
            # Post-check
            if migration.post_check_sql:
                check_result = await self.db.fetch_one(migration.post_check_sql)
                if not check_result or not check_result[0]:
                    raise MigrationError("Post-check failed, rolling back")
            
            # Record success
            await self._record_migration(migration)
            
            result.success = True
            result.completed_at = datetime.now(timezone.utc)
            
            logger.info(f"Migration complete: {migration.name}")
            
            return result
            
        except Exception as e:
            logger.error(f"Migration failed: {e}")
            
            # Attempt rollback
            if not dry_run:
                try:
                    await self.db.execute(migration.down_sql)
                    logger.info("Rollback successful")
                except Exception as rollback_error:
                    logger.error(f"Rollback failed: {rollback_error}")
            
            result.success = False
            result.error = str(e)
            
            return result
            
        finally:
            if migration.requires_lock:
                await self.locks.release("migration_lock")


# =============================================================================
# EXAMPLE: RENAME COLUMN MIGRATION SET
# =============================================================================

RENAME_COLUMN_MIGRATIONS = [
    # Phase 1: Expand - Add new column
    Migration(
        version="2024_01_15_001",
        name="add_full_name_column",
        phase=MigrationPhase.EXPAND,
        up_sql="""
            ALTER TABLE users 
            ADD COLUMN full_name VARCHAR(255);
            
            -- Trigger to keep columns in sync during transition
            CREATE OR REPLACE FUNCTION sync_name_columns()
            RETURNS TRIGGER AS $$
            BEGIN
                IF TG_OP = 'INSERT' THEN
                    -- Old code inserts only name; new code may set both
                    NEW.full_name := COALESCE(NEW.full_name, NEW.name);
                    NEW.name := COALESCE(NEW.name, NEW.full_name);
                ELSIF NEW.name IS DISTINCT FROM OLD.name THEN
                    NEW.full_name := NEW.name;
                ELSIF NEW.full_name IS DISTINCT FROM OLD.full_name THEN
                    NEW.name := NEW.full_name;
                END IF;
                RETURN NEW;
            END;
            $$ LANGUAGE plpgsql;
            
            CREATE TRIGGER sync_names
            BEFORE INSERT OR UPDATE ON users
            FOR EACH ROW EXECUTE FUNCTION sync_name_columns();
        """,
        down_sql="""
            DROP TRIGGER IF EXISTS sync_names ON users;
            DROP FUNCTION IF EXISTS sync_name_columns();
            ALTER TABLE users DROP COLUMN IF EXISTS full_name;
        """,
        post_check_sql="""
            SELECT EXISTS (
                SELECT 1 FROM information_schema.columns 
                WHERE table_name = 'users' AND column_name = 'full_name'
            );
        """
    ),
    
    # Phase 2: Migrate - Backfill data
    Migration(
        version="2024_01_16_001",
        name="backfill_full_name",
        phase=MigrationPhase.MIGRATE,
        up_sql="""
            -- Backfill in batches to avoid locking
            DO $$
            DECLARE
                batch_size INT := 10000;
                updated INT;
            BEGIN
                LOOP
                    UPDATE users 
                    SET full_name = name 
                    WHERE full_name IS NULL 
                    AND id IN (
                        SELECT id FROM users 
                        WHERE full_name IS NULL 
                        LIMIT batch_size
                    );
                    
                    GET DIAGNOSTICS updated = ROW_COUNT;
                    EXIT WHEN updated = 0;
                    
                    -- Commit each batch so its row locks are released
                    -- (PostgreSQL 11+; the DO block must not run inside
                    -- an explicit transaction, or it is all one transaction)
                    COMMIT;
                    
                    -- Small pause between batches
                    PERFORM pg_sleep(0.1);
                END LOOP;
            END $$;
        """,
        down_sql="-- No rollback needed for backfill",
        post_check_sql="""
            SELECT COUNT(*) = 0 
            FROM users 
            WHERE full_name IS NULL AND name IS NOT NULL;
        """,
        estimated_duration="30 minutes for 1M rows"
    ),
    
    # Phase 3: Contract - Remove old column (after code fully migrated)
    Migration(
        version="2024_01_20_001",
        name="remove_name_column",
        phase=MigrationPhase.CONTRACT,
        up_sql="""
            DROP TRIGGER IF EXISTS sync_names ON users;
            DROP FUNCTION IF EXISTS sync_name_columns();
            ALTER TABLE users DROP COLUMN name;
        """,
        down_sql="""
            ALTER TABLE users ADD COLUMN name VARCHAR(255);
            UPDATE users SET name = full_name;
        """,
        pre_check_sql="""
            -- Ensure no code is reading from 'name' column
            -- This should be checked via application logs/metrics
            SELECT true;
        """,
        requires_lock=True
    ),
]

# deployment/auto_rollback.py

"""
Automated rollback based on metrics.

If deployment causes problems, automatically roll back.
"""

from dataclasses import dataclass
from typing import Optional
import asyncio
import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)


@dataclass
class RollbackConfig:
    """Configuration for automated rollback."""
    # Metric thresholds
    max_error_rate: float = 0.05  # 5%
    max_latency_p99_ms: int = 1000  # 1 second
    min_success_rate: float = 0.95  # 95%
    
    # Timing
    observation_window_seconds: int = 300  # 5 minutes
    check_interval_seconds: int = 30
    
    # Behavior
    auto_rollback_enabled: bool = True
    require_confirmation: bool = False  # For critical services


class AutomatedRollbackMonitor:
    """
    Monitors deployments and triggers rollback if metrics degrade.
    """
    
    def __init__(
        self,
        metrics_client,
        deployer,
        alerter,
        config: RollbackConfig = None
    ):
        self.metrics = metrics_client
        self.deployer = deployer
        self.alerter = alerter
        self.config = config or RollbackConfig()
    
    async def monitor_deployment(
        self,
        deployment_id: str,
        baseline_metrics: dict
    ):
        """
        Monitor a deployment and rollback if needed.
        """
        logger.info(f"Starting rollback monitor for deployment {deployment_id}")
        
        start_time = datetime.now(timezone.utc)
        end_time = start_time + timedelta(seconds=self.config.observation_window_seconds)
        
        while datetime.now(timezone.utc) < end_time:
            await asyncio.sleep(self.config.check_interval_seconds)
            
            # Check metrics
            current_metrics = await self._get_current_metrics()
            
            # Compare to baseline and thresholds
            should_rollback, reason = self._should_rollback(
                baseline_metrics,
                current_metrics
            )
            
            if should_rollback:
                logger.warning(f"Rollback triggered: {reason}")
                
                if self.config.require_confirmation:
                    # Alert and wait for confirmation
                    await self.alerter.send_critical(
                        f"Deployment {deployment_id} may need rollback: {reason}. "
                        f"Reply 'ROLLBACK' to confirm."
                    )
                    # In a real system, this would wait for response
                elif self.config.auto_rollback_enabled:
                    await self._execute_rollback(deployment_id, reason)
                    return RollbackResult(
                        rolled_back=True,
                        reason=reason,
                        metrics=current_metrics
                    )
        
        logger.info(f"Deployment {deployment_id} passed monitoring period")
        return RollbackResult(rolled_back=False, metrics=current_metrics)
    
    def _should_rollback(
        self,
        baseline: dict,
        current: dict
    ) -> tuple[bool, Optional[str]]:
        """
        Determine if rollback is needed.
        """
        # Check absolute thresholds
        if current['error_rate'] > self.config.max_error_rate:
            return True, f"Error rate {current['error_rate']:.2%} exceeds threshold"
        
        if current['latency_p99_ms'] > self.config.max_latency_p99_ms:
            return True, f"Latency {current['latency_p99_ms']}ms exceeds threshold"
        
        if current['success_rate'] < self.config.min_success_rate:
            return True, f"Success rate {current['success_rate']:.2%} below threshold"
        
        # Check relative degradation vs baseline
        if baseline:
            error_increase = current['error_rate'] - baseline['error_rate']
            if error_increase > 0.02:  # 2% absolute increase
                return True, f"Error rate increased by {error_increase:.2%}"
            
            latency_increase_pct = (
                (current['latency_p99_ms'] - baseline['latency_p99_ms']) /
                baseline['latency_p99_ms']
            ) if baseline['latency_p99_ms'] > 0 else 0
            
            if latency_increase_pct > 0.5:  # 50% increase
                return True, f"Latency increased by {latency_increase_pct:.0%}"
        
        return False, None
    
    async def _execute_rollback(self, deployment_id: str, reason: str):
        """Execute the rollback."""
        logger.warning(f"Executing rollback for {deployment_id}")
        
        # Send alert
        await self.alerter.send_critical(
            f"AUTO-ROLLBACK executed for deployment {deployment_id}. "
            f"Reason: {reason}"
        )
        
        # Execute rollback
        await self.deployer.rollback(deployment_id)
        
        logger.info(f"Rollback complete for {deployment_id}")