feat: persist probe quality and notification outbox

backend/migrations/versions/20260803_reliability_monitoring.py

@@ -0,0 +1,103 @@
+"""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 Base
+
+
+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",
+}
+
+
+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 in Base.metadata.sorted_tables:
+        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 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"
+                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)
+    for table in Base.metadata.sorted_tables:
+        existing_indexes = {
+            index["name"] for index in inspector.get_indexes(table.name)
+        }
+        for index in table.indexes:
+            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":
+        connection.execute(
+            text("ALTER TYPE alerttypeenum 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 absent tables,
+    columns, and indexes, then writes one immutable revision marker. It never
+    issues DROP, DELETE, UPDATE, or data-copy statements.
+    """
+    _migration_metadata.create_all(bind=connection)
+    Base.metadata.create_all(bind=connection)
+
+    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))