"""Add durable reliability-monitoring persistence without destructive DDL.""" from datetime import datetime from sqlalchemy import Column, DateTime, MetaData, String, Table, inspect, select, text from sqlalchemy.engine import Connection from app.models import AlertEvent, Device, NotificationOutbox, PingRecord REVISION = "20260803_reliability_monitoring" _migration_metadata = MetaData() _schema_migrations = Table( "schema_migrations", _migration_metadata, Column("revision", String(64), primary_key=True), Column("applied_at", DateTime, nullable=False, default=datetime.now), ) _DEFAULTS_FOR_EXISTING_ROWS = { "is_valid": "TRUE", "notification_attempts": "0", } _REVISION_COLUMNS = { "devices": ( Device.__table__, { "probe_packets_per_round", "offline_consecutive_rounds", "degraded_window_rounds", "degraded_loss_percent", "recovery_consecutive_clean_rounds", }, ), "ping_records": ( PingRecord.__table__, { "sent_count", "received_count", "packet_loss_percent", "average_rtt_ms", "is_valid", "failure_reason", }, ), "alert_events": ( AlertEvent.__table__, { "notification_attempts", "last_notification_error", "related_event_id", "previous_status", "current_status", }, ), } def _quote(connection: Connection, identifier: str) -> str: """Quote one database identifier through the active dialect.""" return connection.dialect.identifier_preparer.quote(identifier) def _add_missing_columns(connection: Connection) -> None: """Add only newly required nullable/defaulted columns to existing tables.""" inspector = inspect(connection) existing_tables = set(inspector.get_table_names()) for table_name, (table, column_names) in _REVISION_COLUMNS.items(): if table_name not in existing_tables: continue existing_columns = { column["name"] for column in inspector.get_columns(table_name) } for column in table.columns: if column.name not in column_names or column.name in existing_columns: continue column_type = connection.dialect.type_compiler.process(column.type) definition = f"{_quote(connection, column.name)} {column_type}" default = _DEFAULTS_FOR_EXISTING_ROWS.get(column.name) if default is not None: if connection.dialect.name == "sqlite" and default == "TRUE": default = "1" if column.name == "is_valid": definition = f"{definition} NOT NULL DEFAULT {default}" else: definition = f"{definition} DEFAULT {default}" connection.execute( text( f"ALTER TABLE {_quote(connection, table_name)} " f"ADD COLUMN {definition}" ) ) def _create_missing_indexes(connection: Connection) -> None: """Create new performance indexes for tables that predate this revision.""" inspector = inspect(connection) table = PingRecord.__table__ if table.name not in inspector.get_table_names(): return existing_indexes = { index["name"] for index in inspector.get_indexes(table.name) } index = next( item for item in table.indexes if item.name == "idx_ping_records_device_created_at" ) if index.name not in existing_indexes: index.create(bind=connection) def _upgrade_postgresql_alert_type(connection: Connection) -> None: """Permit the degraded alert value for databases using PostgreSQL enums.""" if connection.dialect.name == "postgresql": enum_name = AlertEvent.__table__.c.alert_type.type.name enum_exists = connection.execute( text("SELECT 1 FROM pg_type WHERE typname = :enum_name"), {"enum_name": enum_name}, ).scalar_one_or_none() if enum_exists is None: return connection.execute( text( f"ALTER TYPE {_quote(connection, enum_name)} " "ADD VALUE IF NOT EXISTS 'degraded'" ) ) def run_reliability_migration(connection: Connection) -> None: """Apply this revision once while retaining all existing monitoring data. This runner is deliberately idempotent: it creates only this revision's outbox table, missing revision columns, and indexes, then writes one immutable revision marker. It never issues DROP, DELETE, UPDATE, or data-copy statements. The application owns global metadata creation after this function returns. """ _migration_metadata.create_all(bind=connection) NotificationOutbox.__table__.create(bind=connection, checkfirst=True) already_applied = connection.execute( select(_schema_migrations.c.revision).where( _schema_migrations.c.revision == REVISION ) ).scalar_one_or_none() if already_applied is not None: return _upgrade_postgresql_alert_type(connection) _add_missing_columns(connection) _create_missing_indexes(connection) connection.execute(_schema_migrations.insert().values(revision=REVISION))