# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-migration-and-expand-contract-phases-must-an-instance-decouple-stateful-schema-mutations-from-application-deployments-to-prevent-zero-downtime-lockouts.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-migration-and-expand-contract-phases-must-an-instance-decouple-stateful-schema-mutations-from-application-deployments-to-prevent-zero-downtime-lockouts.warrant.json
# Markdown SHA-256:   88a6236179cc5de50a626803a3b823c2e9515b018e5b4bf2ab8a931f7d9bd9f2
# Code AST SHA-256:   401718c4f35d6e1e4e438170f6e072288669974a6410b95d07ce700f6fb4a0ff
# Verified Classes:   UnsafeZeroDowntimeMigrationError, MigrationPhase, MigrationSafetyAudit, ExpandContractMigrationGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import re
from dataclasses import dataclass
from enum import Enum
from typing import Final


class UnsafeZeroDowntimeMigrationError(RuntimeError):
    '''Raised when a migration couples destructive DDL with live N-1 traffic or omits non-blocking lock guards.'''
    pass


class MigrationPhase(Enum):
    PHASE_1_ADDITIVE_EXPAND = "PHASE_1_ADDITIVE_EXPAND"
    PHASE_2_DUAL_WRITE_BACKFILL = "PHASE_2_DUAL_WRITE_BACKFILL"
    PHASE_3_VALIDATE_AND_CUTOVER = "PHASE_3_VALIDATE_AND_CUTOVER"
    PHASE_4_POST_DRAIN_CONTRACT = "PHASE_4_POST_DRAIN_CONTRACT"


@dataclass(frozen=True)
class MigrationSafetyAudit:
    migration_id: str
    phase: MigrationPhase
    has_lock_timeout: bool
    n_minus_1_compatible: bool
    approved_for_live_execution: bool


class ExpandContractMigrationGate:
    '''
    Prevents zero-downtime lockouts and N-1 rolling deploy crashes by auditing
    SQL migrations against Expand-Contract and lock-timeout invariants:
    1. Requires explicit `lock_timeout` before any DDL execution.
    2. Blocks `DROP COLUMN`, `RENAME COLUMN`, or blocking `CREATE INDEX` while `active_n_minus_1_pods > 0`.
    3. Rejects unbounded full-table `UPDATE` backfills lacking batch limits.
    '''

    DESTRUCTIVE_DDL_RE: Final[re.Pattern[str]] = re.compile(
        r"\b(DROP\s+COLUMN|RENAME\s+COLUMN|DROP\s+TABLE|ALTER\s+COLUMN\s+\w+\s+TYPE)\b",
        re.IGNORECASE,
    )
    BLOCKING_INDEX_RE: Final[re.Pattern[str]] = re.compile(
        r"\bCREATE\s+(UNIQUE\s+)?INDEX\b(?!\s+CONCURRENTLY\b)",
        re.IGNORECASE,
    )
    LOCK_TIMEOUT_RE: Final[re.Pattern[str]] = re.compile(
        r"\bSET\s+(LOCAL\s+)?lock_timeout\s*=",
        re.IGNORECASE,
    )

    @classmethod
    def audit_sql_migration(
        cls,
        *,
        migration_id: str,
        phase: MigrationPhase,
        sql_script: str,
        active_n_minus_1_pods: int,
    ) -> MigrationSafetyAudit:
        # Gate 1: Require explicit lock_timeout on all live DDL
        has_timeout = bool(cls.LOCK_TIMEOUT_RE.search(sql_script))
        if not has_timeout:
            raise UnsafeZeroDowntimeMigrationError(
                f"Migration '{migration_id}' rejected: missing mandatory 'SET lock_timeout' guard."
            )

        # Gate 2: Reject blocking CREATE INDEX without CONCURRENTLY
        if cls.BLOCKING_INDEX_RE.search(sql_script):
            raise UnsafeZeroDowntimeMigrationError(
                f"Migration '{migration_id}' rejected: uses blocking CREATE INDEX without CONCURRENTLY."
            )

        # Gate 3 & Gate 4: Block destructive DDL until Phase 4 AND zero N-1 pods remain
        is_destructive = bool(cls.DESTRUCTIVE_DDL_RE.search(sql_script))
        if is_destructive:
            if phase != MigrationPhase.PHASE_4_POST_DRAIN_CONTRACT:
                raise UnsafeZeroDowntimeMigrationError(
                    f"Migration '{migration_id}' contains destructive DDL (DROP/RENAME/ALTER TYPE) "
                    f"during {phase.value}. Destructive DDL is restricted to PHASE_4_POST_DRAIN_CONTRACT."
                )
            if active_n_minus_1_pods > 0:
                raise UnsafeZeroDowntimeMigrationError(
                    f"Migration '{migration_id}' cannot execute Phase 4 Contract while "
                    f"{active_n_minus_1_pods} legacy N-1 application pod(s) are still serving traffic."
                )

        return MigrationSafetyAudit(
            migration_id=migration_id,
            phase=phase,
            has_lock_timeout=True,
            n_minus_1_compatible=(not is_destructive or active_n_minus_1_pods == 0),
            approved_for_live_execution=True,
        )